You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2021/10/30 15:42:44 UTC
[pulsar] branch branch-2.8 updated (aa09ac4 -> 27e121f)
This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a change to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git.
from aa09ac4 Fix the batch message ack for WebSocket proxy. (#12530)
new 98a5bca Fixed used after recycle issue in OpAddEntry (#12103)
new c36b9d0 [C++] Fixed connection read error logging (#12492)
new 04e8d7e [Broker] Fix messageDedup delete inactive producer name (#12493)
new 1fc1830 Update Producer stats on producer close() (#12500)
new c1e7f94 [Python Client] Python client support using custom Avro schema definition (#12516)
new 27e121f Websocket should pass the encryption context to the consumers (#12539)
The 6 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/bookkeeper/mledger/impl/OpAddEntry.java | 13 ++-
.../service/persistent/MessageDeduplication.java | 2 +
.../service/persistent/MessageDuplicationTest.java | 66 +++++++++++
.../websocket/proxy/ProxyPublishConsumeTest.java | 62 ++++++++++
.../websocket/proxy/SimpleConsumerSocket.java | 3 +
pulsar-client-cpp/lib/ClientConnection.cc | 4 +-
pulsar-client-cpp/python/examples/company.avsc | 19 +++
.../python/pulsar/schema/schema_avro.py | 29 +++--
pulsar-client-cpp/python/schema_test.py | 128 ++++++++++++++++++++-
.../client/impl/ProducerStatsRecorderImpl.java | 90 ++++++++-------
.../client/impl/PartitionedProducerImplTest.java | 33 ++++++
.../client/impl/ProducerStatsRecorderImplTest.java | 20 ++++
.../apache/pulsar/websocket/ConsumerHandler.java | 1 +
.../pulsar/websocket/data/ConsumerMessage.java | 3 +
14 files changed, 408 insertions(+), 65 deletions(-)
create mode 100644 pulsar-client-cpp/python/examples/company.avsc
[pulsar] 03/06: [Broker] Fix messageDedup delete inactive producer
name (#12493)
Posted by mm...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 04e8d7e58774e299c28435caede1ebd7f41a789d
Author: congbo <39...@users.noreply.github.com>
AuthorDate: Wed Oct 27 10:01:39 2021 +0800
[Broker] Fix messageDedup delete inactive producer name (#12493)
Now, remove inactive producerName in MessageDeduplication when producer close. But the producer has been closed before topic unload, this producerName will not be remove if producer don't connect broker with the same producerName.
When topic recover `MessageDeduplication`, we should put every producerName into inactive producerNameMap. When producer with the same name, we will remove it from the inactive map, if this producerName can not connect within brokerDeduplicationProducerInactivityTimeoutMinutes, we can remove it.
---
.../service/persistent/MessageDeduplication.java | 2 +
.../service/persistent/MessageDuplicationTest.java | 66 ++++++++++++++++++++++
2 files changed, 68 insertions(+)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java
index 6aba2c4..1c25de0 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java
@@ -141,6 +141,7 @@ public class MessageDeduplication {
private CompletableFuture<Void> recoverSequenceIdsMap() {
// Load the sequence ids from the snapshot in the cursor properties
managedCursor.getProperties().forEach((k, v) -> {
+ inactiveProducers.put(k, System.currentTimeMillis());
highestSequencedPushed.put(k, v);
highestSequencedPersisted.put(k, v);
});
@@ -171,6 +172,7 @@ public class MessageDeduplication {
long sequenceId = Math.max(md.getHighestSequenceId(), md.getSequenceId());
highestSequencedPushed.put(producerName, sequenceId);
highestSequencedPersisted.put(producerName, sequenceId);
+ inactiveProducers.put(producerName, System.currentTimeMillis());
entry.release();
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/MessageDuplicationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/MessageDuplicationTest.java
index ed3e2bb..1777a0a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/MessageDuplicationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/MessageDuplicationTest.java
@@ -32,7 +32,13 @@ import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.protocol.Commands;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
+import org.awaitility.Awaitility;
import org.testng.annotations.Test;
+
+import java.lang.reflect.Field;
+import java.util.Map;
+
import static org.apache.pulsar.common.protocol.Commands.serializeMetadataAndPayload;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
@@ -42,7 +48,9 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
@@ -146,6 +154,64 @@ public class MessageDuplicationTest {
}
@Test
+ public void testInactiveProducerRemove() throws Exception {
+ PulsarService pulsarService = mock(PulsarService.class);
+ PersistentTopic topic = mock(PersistentTopic.class);
+ ManagedLedger managedLedger = mock(ManagedLedger.class);
+
+ ServiceConfiguration serviceConfiguration = new ServiceConfiguration();
+ serviceConfiguration.setBrokerDeduplicationEntriesInterval(BROKER_DEDUPLICATION_ENTRIES_INTERVAL);
+ serviceConfiguration.setBrokerDeduplicationMaxNumberOfProducers(BROKER_DEDUPLICATION_MAX_NUMBER_PRODUCERS);
+ serviceConfiguration.setReplicatorPrefix(REPLICATOR_PREFIX);
+ serviceConfiguration.setBrokerDeduplicationProducerInactivityTimeoutMinutes(1);
+
+ doReturn(serviceConfiguration).when(pulsarService).getConfiguration();
+ MessageDeduplication messageDeduplication = spy(new MessageDeduplication(pulsarService, topic, managedLedger));
+ doReturn(true).when(messageDeduplication).isEnabled();
+
+ Topic.PublishContext publishContext = mock(Topic.PublishContext.class);
+
+ Field field = MessageDeduplication.class.getDeclaredField("inactiveProducers");
+ field.setAccessible(true);
+ Map<String, Long> map = (Map<String, Long>) field.get(messageDeduplication);
+
+ String producerName1 = "test1";
+ when(publishContext.getHighestSequenceId()).thenReturn(2L);
+ when(publishContext.getSequenceId()).thenReturn(1L);
+ when(publishContext.getProducerName()).thenReturn(producerName1);
+ messageDeduplication.isDuplicate(publishContext, null);
+
+ String producerName2 = "test2";
+ when(publishContext.getProducerName()).thenReturn(producerName2);
+ messageDeduplication.isDuplicate(publishContext, null);
+
+ String producerName3 = "test3";
+ when(publishContext.getProducerName()).thenReturn(producerName3);
+ messageDeduplication.isDuplicate(publishContext, null);
+
+ messageDeduplication.producerRemoved(producerName1);
+ assertTrue(map.containsKey(producerName1));
+ messageDeduplication.producerAdded(producerName1);
+ assertFalse(map.containsKey(producerName1));
+ messageDeduplication.purgeInactiveProducers();
+ // messageDeduplication.purgeInactiveProducers() will remove producer2 and producer3
+ map.put(producerName2, System.currentTimeMillis() - 70000);
+ map.put(producerName3, System.currentTimeMillis() - 70000);
+ messageDeduplication.purgeInactiveProducers();
+ assertFalse(map.containsKey(producerName2));
+ assertFalse(map.containsKey(producerName3));
+
+ field = MessageDeduplication.class.getDeclaredField("highestSequencedPushed");
+ field.setAccessible(true);
+ ConcurrentOpenHashMap<String, Long> highestSequencedPushed = (ConcurrentOpenHashMap<String, Long>) field.get(messageDeduplication);
+
+ assertEquals((long) highestSequencedPushed.get(producerName1), 2L);
+ assertFalse(highestSequencedPushed.containsKey(producerName2));
+ assertFalse(highestSequencedPushed.containsKey(producerName3));
+
+ }
+
+ @Test
public void testIsDuplicateWithFailure() {
PulsarService pulsarService = mock(PulsarService.class);
[pulsar] 05/06: [Python Client] Python client support using custom
Avro schema definition (#12516)
Posted by mm...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c1e7f940d9d6a408a3a2e0745eb4b6d943bd3920
Author: ran <ga...@126.com>
AuthorDate: Sat Oct 30 08:17:45 2021 +0800
[Python Client] Python client support using custom Avro schema definition (#12516)
### Motivation
Currently, the Python client didn't support using schema definition to generate `AvroSchema`, so users couldn't use the schema definition file in the Python client.
### Modifications
Add a new init-param `schema_definition` for `AvroSchema` to support initializing the `AvroSchema` by an Avro schema definition.
```
class AvroSchema(Schema):
def __init__(self, record_cls, schema_definition=None):
if record_cls is None and schema_definition is None:
raise AssertionError("The param record_cls and schema_definition shouldn't be both None.")
if record_cls is not None:
self._schema = record_cls.schema()
else:
self._schema = schema_definition
super(AvroSchema, self).__init__(record_cls, _pulsar.SchemaType.AVRO, self._schema, 'AVRO')
```
### How to use
Assume that there is a company Avro schema definition file `company.avsc` like this.
```
{
"doc": "this is doc",
"namespace": "example.avro",
"type": "record",
"name": "Company",
"fields": [
{"name": "name", "type": ["null", "string"]},
{"name": "address", "type": ["null", "string"]},
{"name": "employees", "type": ["null", {"type": "array", "items": {
"type": "record",
"name": "Employee",
"fields": [
{"name": "name", "type": ["null", "string"]},
{"name": "age", "type": ["null", "int"]}
]
}}]},
{"name": "labels", "type": ["null", {"type": "map", "values": "string"}]}
]
}
```
Users could load schema from file by `avro.schema` or `fastavro.schema`
> refer to [load_schema](https://fastavro.readthedocs.io/en/latest/schema.html#fastavro._schema_py.load_schema) or [Avro Schema](http://avro.apache.org/docs/current/gettingstartedpython.html)
```
schema_definition = load_schema("examples/company.avsc")
# schema_definition = avro.schema.parse(open("examples/company.avsc", "rb").read()).to_json()
avro_schema = AvroSchema(None, schema_definition=schema_definition)
producer = client.create_producer(
topic=topic,
schema=avro_schema)
consumer = client.subscribe(topic, 'test', schema=avro_schema)
company = {
"name": "company-name" + str(i),
"address": 'xxx road xxx street ' + str(i),
"employees": [
{"name": "user" + str(i), "age": 20 + i},
{"name": "user" + str(i), "age": 30 + i},
{"name": "user" + str(i), "age": 35 + i},
],
"labels": {
"industry": "software" + str(i),
"scale": ">100",
"funds": "1000000.0"
}
}
producer.send(company)
msg = consumer.receive()
# Users could get a dict object by `value()` method.
msg.value()
```
---
pulsar-client-cpp/python/examples/company.avsc | 19 +++
.../python/pulsar/schema/schema_avro.py | 29 +++--
pulsar-client-cpp/python/schema_test.py | 128 ++++++++++++++++++++-
3 files changed, 162 insertions(+), 14 deletions(-)
diff --git a/pulsar-client-cpp/python/examples/company.avsc b/pulsar-client-cpp/python/examples/company.avsc
new file mode 100644
index 0000000..cdb595f
--- /dev/null
+++ b/pulsar-client-cpp/python/examples/company.avsc
@@ -0,0 +1,19 @@
+{
+ "doc": "this is doc",
+ "namespace": "example.avro",
+ "type": "record",
+ "name": "Company",
+ "fields": [
+ {"name": "name", "type": ["null", "string"]},
+ {"name": "address", "type": ["null", "string"]},
+ {"name": "employees", "type": ["null", {"type": "array", "items": {
+ "type": "record",
+ "name": "Employee",
+ "fields": [
+ {"name": "name", "type": ["null", "string"]},
+ {"name": "age", "type": ["null", "int"]}
+ ]
+ }}]},
+ {"name": "labels", "type": ["null", {"type": "map", "values": "string"}]}
+ ]
+}
\ No newline at end of file
diff --git a/pulsar-client-cpp/python/pulsar/schema/schema_avro.py b/pulsar-client-cpp/python/pulsar/schema/schema_avro.py
index e76fc51..5861505 100644
--- a/pulsar-client-cpp/python/pulsar/schema/schema_avro.py
+++ b/pulsar-client-cpp/python/pulsar/schema/schema_avro.py
@@ -32,10 +32,15 @@ except ModuleNotFoundError:
if HAS_AVRO:
class AvroSchema(Schema):
- def __init__(self, record_cls):
- super(AvroSchema, self).__init__(record_cls, _pulsar.SchemaType.AVRO,
- record_cls.schema(), 'AVRO')
- self._schema = record_cls.schema()
+ def __init__(self, record_cls, schema_definition=None):
+ if record_cls is None and schema_definition is None:
+ raise AssertionError("The param record_cls and schema_definition shouldn't be both None.")
+
+ if record_cls is not None:
+ self._schema = record_cls.schema()
+ else:
+ self._schema = schema_definition
+ super(AvroSchema, self).__init__(record_cls, _pulsar.SchemaType.AVRO, self._schema, 'AVRO')
def _get_serialized_value(self, x):
if isinstance(x, enum.Enum):
@@ -53,9 +58,14 @@ if HAS_AVRO:
return x
def encode(self, obj):
- self._validate_object_type(obj)
buffer = io.BytesIO()
- m = self.encode_dict(obj.__dict__)
+ m = obj
+ if self._record_cls is not None:
+ self._validate_object_type(obj)
+ m = self.encode_dict(obj.__dict__)
+ elif not isinstance(obj, dict):
+ raise ValueError('If using the custom schema, the record data should be dict type.')
+
fastavro.schemaless_writer(buffer, self._schema, m)
return buffer.getvalue()
@@ -68,11 +78,14 @@ if HAS_AVRO:
def decode(self, data):
buffer = io.BytesIO(data)
d = fastavro.schemaless_reader(buffer, self._schema)
- return self._record_cls(**d)
+ if self._record_cls is not None:
+ return self._record_cls(**d)
+ else:
+ return d
else:
class AvroSchema(Schema):
- def __init__(self, _record_cls):
+ def __init__(self, _record_cls, _schema_definition):
raise Exception("Avro library support was not found. Make sure to install Pulsar client " +
"with Avro support: pip3 install 'pulsar-client[avro]'")
diff --git a/pulsar-client-cpp/python/schema_test.py b/pulsar-client-cpp/python/schema_test.py
index 7adbcbe..d2554da 100755
--- a/pulsar-client-cpp/python/schema_test.py
+++ b/pulsar-client-cpp/python/schema_test.py
@@ -25,6 +25,7 @@ import pulsar
from pulsar.schema import *
from enum import Enum
import json
+from fastavro.schema import load_schema
class SchemaTest(TestCase):
@@ -1145,12 +1146,127 @@ class SchemaTest(TestCase):
client.close()
- def test(self):
- class NamespaceDemo(Record):
- _namespace = 'xxx.xxx.xxx'
- x = String()
- y = Integer()
- print('schema: ', NamespaceDemo.schema())
+ def custom_schema_test(self):
+
+ def encode_and_decode(schema_definition):
+ avro_schema = AvroSchema(None, schema_definition=schema_definition)
+
+ company = {
+ "name": "company-name",
+ "address": 'xxx road xxx street',
+ "employees": [
+ {"name": "user1", "age": 25},
+ {"name": "user2", "age": 30},
+ {"name": "user3", "age": 35},
+ ],
+ "labels": {
+ "industry": "software",
+ "scale": ">100",
+ "funds": "1000000.0"
+ }
+ }
+ data = avro_schema.encode(company)
+ company_decode = avro_schema.decode(data)
+ self.assertEqual(company, company_decode)
+
+ schema_definition = {
+ 'doc': 'this is doc',
+ 'namespace': 'example.avro',
+ 'type': 'record',
+ 'name': 'Company',
+ 'fields': [
+ {'name': 'name', 'type': ['null', 'string']},
+ {'name': 'address', 'type': ['null', 'string']},
+ {'name': 'employees', 'type': ['null', {'type': 'array', 'items': {
+ 'type': 'record',
+ 'name': 'Employee',
+ 'fields': [
+ {'name': 'name', 'type': ['null', 'string']},
+ {'name': 'age', 'type': ['null', 'int']}
+ ]
+ }}]},
+ {'name': 'labels', 'type': ['null', {'type': 'map', 'values': 'string'}]}
+ ]
+ }
+ encode_and_decode(schema_definition)
+ # Users could load schema from file by `fastavro.schema`
+ # Or use `avro.schema` like this `avro.schema.parse(open("examples/company.avsc", "rb").read()).to_json()`
+ encode_and_decode(load_schema("examples/company.avsc"))
+
+ def custom_schema_produce_and_consume_test(self):
+ client = pulsar.Client(self.serviceUrl)
+
+ def produce_and_consume(topic, schema_definition):
+ print('custom schema produce and consume test topic - ', topic)
+ example_avro_schema = AvroSchema(None, schema_definition=schema_definition)
+
+ producer = client.create_producer(
+ topic=topic,
+ schema=example_avro_schema)
+ consumer = client.subscribe(topic, 'test', schema=example_avro_schema)
+
+ for i in range(0, 10):
+ company = {
+ "name": "company-name" + str(i),
+ "address": 'xxx road xxx street ' + str(i),
+ "employees": [
+ {"name": "user" + str(i), "age": 20 + i},
+ {"name": "user" + str(i), "age": 30 + i},
+ {"name": "user" + str(i), "age": 35 + i},
+ ],
+ "labels": {
+ "industry": "software" + str(i),
+ "scale": ">100",
+ "funds": "1000000.0"
+ }
+ }
+ producer.send(company)
+
+ for i in range(0, 10):
+ msg = consumer.receive()
+ company = {
+ "name": "company-name" + str(i),
+ "address": 'xxx road xxx street ' + str(i),
+ "employees": [
+ {"name": "user" + str(i), "age": 20 + i},
+ {"name": "user" + str(i), "age": 30 + i},
+ {"name": "user" + str(i), "age": 35 + i},
+ ],
+ "labels": {
+ "industry": "software" + str(i),
+ "scale": ">100",
+ "funds": "1000000.0"
+ }
+ }
+ self.assertEqual(msg.value(), company)
+ consumer.acknowledge(msg)
+
+ consumer.close()
+ producer.close()
+
+ schema_definition = {
+ 'doc': 'this is doc',
+ 'namespace': 'example.avro',
+ 'type': 'record',
+ 'name': 'Company',
+ 'fields': [
+ {'name': 'name', 'type': ['null', 'string']},
+ {'name': 'address', 'type': ['null', 'string']},
+ {'name': 'employees', 'type': ['null', {'type': 'array', 'items': {
+ 'type': 'record',
+ 'name': 'Employee',
+ 'fields': [
+ {'name': 'name', 'type': ['null', 'string']},
+ {'name': 'age', 'type': ['null', 'int']}
+ ]
+ }}]},
+ {'name': 'labels', 'type': ['null', {'type': 'map', 'values': 'string'}]}
+ ]
+ }
+ produce_and_consume('custom-schema-test-1', schema_definition=schema_definition)
+ produce_and_consume('custom-schema-test-2', schema_definition=load_schema("examples/company.avsc"))
+
+ client.close()
if __name__ == '__main__':
main()
[pulsar] 02/06: [C++] Fixed connection read error logging (#12492)
Posted by mm...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c36b9d0337bf956eff0e2da068cfbdb3f5a273bb
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Mon Oct 25 21:15:57 2021 -0700
[C++] Fixed connection read error logging (#12492)
---
pulsar-client-cpp/lib/ClientConnection.cc | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/pulsar-client-cpp/lib/ClientConnection.cc b/pulsar-client-cpp/lib/ClientConnection.cc
index 6e6f2a3..6e5245e 100644
--- a/pulsar-client-cpp/lib/ClientConnection.cc
+++ b/pulsar-client-cpp/lib/ClientConnection.cc
@@ -584,9 +584,9 @@ void ClientConnection::handleRead(const boost::system::error_code& err, size_t b
if (err || bytesTransferred == 0) {
if (err) {
if (err == boost::asio::error::operation_aborted) {
- LOG_DEBUG(cnxString_ << "Read failed: " << err.message());
+ LOG_DEBUG(cnxString_ << "Read operation was canceled: " << err.message());
} else {
- LOG_ERROR(cnxString_ << "Read operation was cancelled");
+ LOG_ERROR(cnxString_ << "Read operation failed: " << err.message());
}
} // else: bytesTransferred == 0, which means server has closed the connection
close();
[pulsar] 04/06: Update Producer stats on producer close() (#12500)
Posted by mm...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 1fc1830321764aeeb014695b919fd39cb1c3d5f7
Author: Kai Wang <wa...@gmail.com>
AuthorDate: Wed Oct 27 00:50:47 2021 +0800
Update Producer stats on producer close() (#12500)
---
.../client/impl/ProducerStatsRecorderImpl.java | 90 +++++++++++-----------
.../client/impl/PartitionedProducerImplTest.java | 33 ++++++++
.../client/impl/ProducerStatsRecorderImplTest.java | 20 +++++
3 files changed, 100 insertions(+), 43 deletions(-)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerStatsRecorderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerStatsRecorderImpl.java
index e7075da..bb420db 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerStatsRecorderImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerStatsRecorderImpl.java
@@ -111,49 +111,7 @@ public class ProducerStatsRecorderImpl implements ProducerStatsRecorder {
}
try {
- long now = System.nanoTime();
- double elapsed = (now - oldTime) / 1e9;
- oldTime = now;
-
- long currentNumMsgsSent = numMsgsSent.sumThenReset();
- long currentNumBytesSent = numBytesSent.sumThenReset();
- long currentNumSendFailedMsgs = numSendFailed.sumThenReset();
- long currentNumAcksReceived = numAcksReceived.sumThenReset();
-
- totalMsgsSent.add(currentNumMsgsSent);
- totalBytesSent.add(currentNumBytesSent);
- totalSendFailed.add(currentNumSendFailedMsgs);
- totalAcksReceived.add(currentNumAcksReceived);
-
- synchronized (ds) {
- latencyPctValues = ds.getQuantiles(PERCENTILES);
- ds.reset();
- }
-
- sendMsgsRate = currentNumMsgsSent / elapsed;
- sendBytesRate = currentNumBytesSent / elapsed;
-
- if ((currentNumMsgsSent | currentNumSendFailedMsgs | currentNumAcksReceived
- | currentNumMsgsSent) != 0) {
-
- for (int i = 0; i < latencyPctValues.length; i++) {
- if (Double.isNaN(latencyPctValues[i])) {
- latencyPctValues[i] = 0;
- }
- }
-
- log.info("[{}] [{}] Pending messages: {} --- Publish throughput: {} msg/s --- {} Mbit/s --- "
- + "Latency: med: {} ms - 95pct: {} ms - 99pct: {} ms - 99.9pct: {} ms - max: {} ms --- "
- + "Ack received rate: {} ack/s --- Failed messages: {}", producer.getTopic(),
- producer.getProducerName(), producer.getPendingQueueSize(),
- THROUGHPUT_FORMAT.format(sendMsgsRate),
- THROUGHPUT_FORMAT.format(sendBytesRate / 1024 / 1024 * 8),
- DEC.format(latencyPctValues[0]), DEC.format(latencyPctValues[2]),
- DEC.format(latencyPctValues[3]), DEC.format(latencyPctValues[4]),
- DEC.format(latencyPctValues[5]),
- THROUGHPUT_FORMAT.format(currentNumAcksReceived / elapsed), currentNumSendFailedMsgs);
- }
-
+ updateStats();
} catch (Exception e) {
log.error("[{}] [{}]: {}", producer.getTopic(), producer.getProducerName(), e.getMessage());
} finally {
@@ -171,6 +129,51 @@ public class ProducerStatsRecorderImpl implements ProducerStatsRecorder {
return statTimeout;
}
+ protected void updateStats() {
+ long now = System.nanoTime();
+ double elapsed = (now - oldTime) / 1e9;
+ oldTime = now;
+
+ long currentNumMsgsSent = numMsgsSent.sumThenReset();
+ long currentNumBytesSent = numBytesSent.sumThenReset();
+ long currentNumSendFailedMsgs = numSendFailed.sumThenReset();
+ long currentNumAcksReceived = numAcksReceived.sumThenReset();
+
+ totalMsgsSent.add(currentNumMsgsSent);
+ totalBytesSent.add(currentNumBytesSent);
+ totalSendFailed.add(currentNumSendFailedMsgs);
+ totalAcksReceived.add(currentNumAcksReceived);
+
+ synchronized (ds) {
+ latencyPctValues = ds.getQuantiles(PERCENTILES);
+ ds.reset();
+ }
+
+ sendMsgsRate = currentNumMsgsSent / elapsed;
+ sendBytesRate = currentNumBytesSent / elapsed;
+
+ if ((currentNumMsgsSent | currentNumSendFailedMsgs | currentNumAcksReceived
+ | currentNumMsgsSent) != 0) {
+
+ for (int i = 0; i < latencyPctValues.length; i++) {
+ if (Double.isNaN(latencyPctValues[i])) {
+ latencyPctValues[i] = 0;
+ }
+ }
+
+ log.info("[{}] [{}] Pending messages: {} --- Publish throughput: {} msg/s --- {} Mbit/s --- "
+ + "Latency: med: {} ms - 95pct: {} ms - 99pct: {} ms - 99.9pct: {} ms - max: {} ms --- "
+ + "Ack received rate: {} ack/s --- Failed messages: {}", producer.getTopic(),
+ producer.getProducerName(), producer.getPendingQueueSize(),
+ THROUGHPUT_FORMAT.format(sendMsgsRate),
+ THROUGHPUT_FORMAT.format(sendBytesRate / 1024 / 1024 * 8),
+ DEC.format(latencyPctValues[0]), DEC.format(latencyPctValues[2]),
+ DEC.format(latencyPctValues[3]), DEC.format(latencyPctValues[4]),
+ DEC.format(latencyPctValues[5]),
+ THROUGHPUT_FORMAT.format(currentNumAcksReceived / elapsed), currentNumSendFailedMsgs);
+ }
+ }
+
@Override
public void updateNumMsgsSent(long numMsgs, long totalMsgsSize) {
numMsgsSent.add(numMsgs);
@@ -297,6 +300,7 @@ public class ProducerStatsRecorderImpl implements ProducerStatsRecorder {
}
public void cancelStatsTimeout() {
+ this.updateStats();
if (statTimeout != null) {
statTimeout.cancel();
statTimeout = null;
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PartitionedProducerImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PartitionedProducerImplTest.java
index d4c6aa4..60ca37a 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PartitionedProducerImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PartitionedProducerImplTest.java
@@ -153,4 +153,37 @@ public class PartitionedProducerImplTest {
impl.getStats();
}
+ @Test
+ public void testGetStatsWithoutArriveUpdateInterval() throws Exception {
+ String topicName = "test-stats-without-arrive-interval";
+ ClientConfigurationData conf = new ClientConfigurationData();
+ conf.setServiceUrl("pulsar://localhost:6650");
+ conf.setStatsIntervalSeconds(100);
+
+ ThreadFactory threadFactory =
+ new DefaultThreadFactory("client-test-stats", Thread.currentThread().isDaemon());
+ EventLoopGroup eventLoopGroup = EventLoopUtil
+ .newEventLoopGroup(conf.getNumIoThreads(), false, threadFactory);
+
+ PulsarClientImpl clientImpl = new PulsarClientImpl(conf, eventLoopGroup);
+
+ ProducerConfigurationData producerConfData = new ProducerConfigurationData();
+ producerConfData.setMessageRoutingMode(MessageRoutingMode.CustomPartition);
+ producerConfData.setCustomMessageRouter(new CustomMessageRouter());
+
+ assertEquals(Long.parseLong("100"), clientImpl.getConfiguration().getStatsIntervalSeconds());
+
+ PartitionedProducerImpl<byte[]> impl = new PartitionedProducerImpl<>(
+ clientImpl, topicName, producerConfData,
+ 1, null, null, null);
+
+ impl.getProducers().get(0).getStats().incrementSendFailed();
+ ProducerStatsRecorderImpl stats = impl.getStats();
+ assertEquals(stats.getTotalSendFailed(), 0);
+ // When close producer, the ProducerStatsRecorder will update stats immediately
+ impl.close();
+ stats = impl.getStats();
+ assertEquals(stats.getTotalSendFailed(), 1);
+ }
+
}
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerStatsRecorderImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerStatsRecorderImplTest.java
index d654158..f6e7f28 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerStatsRecorderImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerStatsRecorderImplTest.java
@@ -54,4 +54,24 @@ public class ProducerStatsRecorderImplTest {
Thread.sleep(1200);
assertEquals(1000.0, recorder.getSendLatencyMillisMax(), 0.5);
}
+
+ @Test
+ public void testGetStatsAndCancelStatsTimeoutWithoutArriveUpdateInterval() {
+ ClientConfigurationData conf = new ClientConfigurationData();
+ conf.setStatsIntervalSeconds(60);
+ PulsarClientImpl client = mock(PulsarClientImpl.class);
+ when(client.getConfiguration()).thenReturn(conf);
+ Timer timer = new HashedWheelTimer();
+ when(client.timer()).thenReturn(timer);
+ ProducerImpl<?> producer = mock(ProducerImpl.class);
+ when(producer.getTopic()).thenReturn("topic-test");
+ when(producer.getProducerName()).thenReturn("producer-test");
+ when(producer.getPendingQueueSize()).thenReturn(1);
+ ProducerConfigurationData producerConfigurationData = new ProducerConfigurationData();
+ ProducerStatsRecorderImpl recorder = new ProducerStatsRecorderImpl(client, producerConfigurationData, producer);
+ long latencyNs = TimeUnit.SECONDS.toNanos(1);
+ recorder.incrementNumAcksReceived(latencyNs);
+ recorder.cancelStatsTimeout();
+ assertEquals(1000.0, recorder.getSendLatencyMillisMax(), 0.5);
+ }
}
[pulsar] 01/06: Fixed used after recycle issue in OpAddEntry
(#12103)
Posted by mm...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 98a5bca30af53f0b99936098ad8a199e8cc8b357
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Wed Sep 22 11:21:27 2021 -0700
Fixed used after recycle issue in OpAddEntry (#12103)
---
.../java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java | 13 +++++++------
1 file changed, 7 insertions(+), 6 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java
index 665b138..9106b4f 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java
@@ -277,18 +277,19 @@ public class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallba
/**
* It handles add failure on the given ledger. it can be triggered when add-entry fails or times out.
*
- * @param ledger
+ * @param lh
*/
- void handleAddFailure(final LedgerHandle ledger) {
+ void handleAddFailure(final LedgerHandle lh) {
// If we get a write error, we will try to create a new ledger and re-submit the pending writes. If the
- // ledger creation fails (persistent bk failure, another instanche owning the ML, ...), then the writes will
+ // ledger creation fails (persistent bk failure, another instance owning the ML, ...), then the writes will
// be marked as failed.
- ml.mbean.recordAddEntryError();
+ ManagedLedgerImpl finalMl = this.ml;
+ finalMl.mbean.recordAddEntryError();
- ml.getExecutor().executeOrdered(ml.getName(), SafeRun.safeRun(() -> {
+ finalMl.getExecutor().executeOrdered(finalMl.getName(), SafeRun.safeRun(() -> {
// Force the creation of a new ledger. Doing it in a background thread to avoid acquiring ML lock
// from a BK callback.
- ml.ledgerClosed(ledger);
+ finalMl.ledgerClosed(lh);
}));
}
[pulsar] 06/06: Websocket should pass the encryption context to the
consumers (#12539)
Posted by mm...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 27e121f2456a4579985279adb2e98604ac114f95
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Fri Oct 29 17:54:27 2021 -0700
Websocket should pass the encryption context to the consumers (#12539)
---
.../websocket/proxy/ProxyPublishConsumeTest.java | 62 ++++++++++++++++++++++
.../websocket/proxy/SimpleConsumerSocket.java | 3 ++
.../apache/pulsar/websocket/ConsumerHandler.java | 1 +
.../pulsar/websocket/data/ConsumerMessage.java | 3 ++
4 files changed, 69 insertions(+)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java
index 3e3cf9a..6386c51 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTest.java
@@ -29,12 +29,14 @@ import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
+import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import java.net.URI;
import java.nio.charset.StandardCharsets;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
@@ -54,9 +56,11 @@ import javax.ws.rs.core.Response;
import lombok.Cleanup;
import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.client.api.MessageCrypto;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerAccessMode;
import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.impl.crypto.MessageCryptoBc;
import org.apache.pulsar.common.policies.data.AutoTopicCreationOverride;
import org.apache.pulsar.common.policies.data.BacklogQuota;
import org.apache.pulsar.common.policies.data.TopicType;
@@ -935,6 +939,64 @@ public class ProxyPublishConsumeTest extends ProducerConsumerBase {
}
}
+ @Test(timeOut = 20000)
+ public void consumeEncryptedMessages() throws Exception {
+ final String subscription = "my-sub";
+ final String topic = "my-property/my-ns/encrypted" + UUID.randomUUID();
+ final String consumerUri = "ws://localhost:" + proxyServer.getListenPortHTTP().get() +
+ "/ws/v2/consumer/persistent/" + topic + "/" + subscription + "?cryptoFailureAction=CONSUME";
+ final int messages = 10;
+
+ WebSocketClient consumerClient = new WebSocketClient();
+ SimpleConsumerSocket consumeSocket = new SimpleConsumerSocket();
+
+
+ final String rsaPublicKeyData = "data:application/x-pem-file;base64,LS0tLS1CRUdJTiBQVUJMSUMgS0VZLS0tLS0KTUlJQklqQU5CZ2txaGtpRzl3MEJBUUVGQUFPQ0FROEFNSUlCQ2dLQ0FRRUF0S1d3Z3FkblRZck9DditqMU1rVApXZlNIMHdDc0haWmNhOXdBVzNxUDR1dWhsQnZuYjEwSmNGZjVaanpQOUJTWEsrdEhtSTh1b04zNjh2RXY2eWhVClJITTR5dVhxekN4enVBd2tRU28zOXJ6WDhQR0M3cWRqQ043TERKM01ucWlCSXJVc1NhRVAxd3JOc0Ixa0krbzkKRVIxZTVPL3VFUEFvdFA5MzNoSFEwSjJoTUVla0hxTDdzQmxKOThoNk5tc2ljRWFVa2FyZGswVE9YcmxrakMrYwpNZDhaYkdTY1BxSTlNMzhibW4zT0x4RlRu [...]
+ final String rsaPrivateKeyData = "data:application/x-pem-file;base64,LS0tLS1CRUdJTiBSU0EgUFJJVkFURSBLRVktLS0tLQpNSUlFb3dJQkFBS0NBUUVBdEtXd2dxZG5UWXJPQ3YrajFNa1RXZlNIMHdDc0haWmNhOXdBVzNxUDR1dWhsQnZuCmIxMEpjRmY1Wmp6UDlCU1hLK3RIbUk4dW9OMzY4dkV2NnloVVJITTR5dVhxekN4enVBd2tRU28zOXJ6WDhQR0MKN3FkakNON0xESjNNbnFpQklyVXNTYUVQMXdyTnNCMWtJK285RVIxZTVPL3VFUEFvdFA5MzNoSFEwSjJoTUVlawpIcUw3c0JsSjk4aDZObXNpY0VhVWthcmRrMFRPWHJsa2pDK2NNZDhaYkdTY1BxSTlNMzhibW4zT0x4RlRuMXZ0Cmhwdm5YTHZDbUc0TSs2eHRZdEQ [...]
+
+ Producer<byte[]> producer = pulsarClient.newProducer()
+ .topic(topic)
+ .enableBatching(false)
+ .defaultCryptoKeyReader(rsaPublicKeyData)
+ .addEncryptionKey("ws-consumer-a")
+ .create();
+
+ try {
+ consumerClient.start();
+ ClientUpgradeRequest consumerRequest = new ClientUpgradeRequest();
+ Future<Session> consumerFuture = consumerClient.connect(consumeSocket, URI.create(consumerUri), consumerRequest);
+
+ assertTrue(consumerFuture.get().isOpen());
+ assertEquals(consumeSocket.getReceivedMessagesCount(), 0);
+
+ for (int i = 0; i < messages; i++) {
+ producer.sendAsync(String.valueOf(i).getBytes(StandardCharsets.UTF_8));
+ }
+
+ producer.flush();
+ consumeSocket.sendPermits(messages);
+ Awaitility.await().untilAsserted(() ->
+ Assert.assertEquals(consumeSocket.getReceivedMessagesCount(), messages));
+
+ for (JsonObject msg : consumeSocket.messages) {
+ assertTrue(msg.has("encryptionContext"));
+ JsonObject encryptionCtx = msg.getAsJsonObject("encryptionContext");
+ JsonObject keys = encryptionCtx.getAsJsonObject("keys");
+ assertTrue(keys.has("ws-consumer-a"));
+
+ assertTrue(keys.getAsJsonObject("ws-consumer-a").has("keyValue"));
+ }
+
+ // The message should not be acked since we only acked 1 message of the batch message
+ Awaitility.await().untilAsserted(() ->
+ Assert.assertEquals(admin.topics().getStats(topic).getSubscriptions()
+ .get(subscription).getMsgBacklog(), 0));
+
+ } finally {
+ stopWebSocketClient(consumerClient);
+ }
+ }
+
private void verifyTopicStat(Client client, String baseUrl, String topic) {
String statUrl = baseUrl + topic + "/stats";
WebTarget webTarget = client.target(statUrl);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleConsumerSocket.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleConsumerSocket.java
index 749bfdcd..b1a9908 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleConsumerSocket.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/SimpleConsumerSocket.java
@@ -44,6 +44,7 @@ public class SimpleConsumerSocket {
private final CountDownLatch closeLatch;
private Session session;
private final ArrayList<String> consumerBuffer;
+ final ArrayList<JsonObject> messages;
private final AtomicInteger receivedMessages = new AtomicInteger();
// Custom message handler to override standard message processing, if it's needed
private SimpleConsumerMessageHandler customMessageHandler;
@@ -51,6 +52,7 @@ public class SimpleConsumerSocket {
public SimpleConsumerSocket() {
this.closeLatch = new CountDownLatch(1);
consumerBuffer = new ArrayList<>();
+ this.messages = new ArrayList<>();
}
public boolean awaitClose(int duration, TimeUnit unit) throws InterruptedException {
@@ -79,6 +81,7 @@ public class SimpleConsumerSocket {
public synchronized void onMessage(String msg) throws JsonParseException, IOException {
receivedMessages.incrementAndGet();
JsonObject message = new Gson().fromJson(msg, JsonObject.class);
+ this.messages.add(message);
if (message.get(X_PULSAR_MESSAGE_ID) != null) {
String messageId = message.get(X_PULSAR_MESSAGE_ID).getAsString();
consumerBuffer.add(messageId);
diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ConsumerHandler.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ConsumerHandler.java
index 54a5a62..0192188 100644
--- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ConsumerHandler.java
+++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ConsumerHandler.java
@@ -160,6 +160,7 @@ public class ConsumerHandler extends AbstractWebSocketHandler {
dm.properties = msg.getProperties();
dm.publishTime = DateFormatter.format(msg.getPublishTime());
dm.redeliveryCount = msg.getRedeliveryCount();
+ dm.encryptionContext = msg.getEncryptionCtx().orElse(null);
if (msg.getEventTime() != 0) {
dm.eventTime = DateFormatter.format(msg.getEventTime());
}
diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/data/ConsumerMessage.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/data/ConsumerMessage.java
index 9660c95..9091a7e 100644
--- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/data/ConsumerMessage.java
+++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/data/ConsumerMessage.java
@@ -22,6 +22,7 @@ import java.util.Map;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.annotation.JsonInclude.Include;
+import org.apache.pulsar.common.api.EncryptionContext;
@JsonInclude(Include.NON_NULL)
public class ConsumerMessage {
@@ -32,5 +33,7 @@ public class ConsumerMessage {
public int redeliveryCount;
public String eventTime;
+ public EncryptionContext encryptionContext;
+
public String key;
}