You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/01/02 08:02:13 UTC
(pulsar) branch branch-3.0 updated (62170e733d1 -> c981972c212)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 62170e733d1 [fix][broker] Continue using the next provider for authentication if one fails (#23797)
new b9ce087bfbb [fix][client] Make DeadLetterPolicy & KeySharedPolicy serializable (#23718)
new f9f5be5daa8 [improve][log] Print ZK path if write to ZK fails due to data being too large to persist (#23652)
new c981972c212 [fix][client] Cannot access message data inside ProducerInterceptor#onSendAcknowledgement (#23791)
The 3 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:
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 6 +--
.../pulsar/broker/service/BatchMessageTest.java | 5 ++-
.../apache/pulsar/client/api/InterceptorsTest.java | 44 +++++++++++++++++++
.../apache/pulsar/client/api/DeadLetterPolicy.java | 4 +-
.../apache/pulsar/client/api/KeySharedPolicy.java | 5 ++-
.../java/org/apache/pulsar/client/api/Range.java | 4 +-
.../apache/pulsar/client/impl/ProducerImpl.java | 22 +++++-----
.../impl/conf/ConsumerConfigurationData.java | 4 +-
.../client/impl/conf/ReaderConfigurationData.java | 2 +-
.../impl/conf/ConsumerConfigurationDataTest.java | 50 ++++++++++++++++++++++
.../pulsar/metadata/impl/ZKMetadataStore.java | 8 +++-
.../pulsar/metadata/impl/batching/MetadataOp.java | 2 +
12 files changed, 134 insertions(+), 22 deletions(-)
(pulsar) 01/03: [fix][client] Make DeadLetterPolicy & KeySharedPolicy serializable (#23718)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit b9ce087bfbb45512e2689df06b852fe0b070d086
Author: Anurag reddy K <62...@users.noreply.github.com>
AuthorDate: Tue Dec 24 12:11:42 2024 +0530
[fix][client] Make DeadLetterPolicy & KeySharedPolicy serializable (#23718)
Co-authored-by: anurag.reddy <an...@flipkart.com>
(cherry picked from commit 14129e352e4bb4e06cbf2c6c581edcdbaf9f6775)
---
.../pulsar/broker/service/BatchMessageTest.java | 5 ++-
.../apache/pulsar/client/api/DeadLetterPolicy.java | 4 +-
.../apache/pulsar/client/api/KeySharedPolicy.java | 5 ++-
.../java/org/apache/pulsar/client/api/Range.java | 4 +-
.../impl/conf/ConsumerConfigurationData.java | 4 +-
.../client/impl/conf/ReaderConfigurationData.java | 2 +-
.../impl/conf/ConsumerConfigurationDataTest.java | 50 ++++++++++++++++++++++
7 files changed, 66 insertions(+), 8 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageTest.java
index a2d80b2ba60..adc4809127f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageTest.java
@@ -990,8 +990,8 @@ public class BatchMessageTest extends BrokerTestBase {
int numMsgs = 1000;
int batchMessages = 10;
- final String topicName = "persistent://prop/ns-abc/testRetrieveSequenceIdSpecify-" + UUID.randomUUID();
- final String subscriptionName = "sub-1";
+ final String topicName = "persistent://prop/ns-abc/testBatchMessageDispatchingAccordingToPermits-" + UUID.randomUUID();
+ final String subscriptionName = "bmdap-sub-1";
ConsumerImpl<byte[]> consumer1 = (ConsumerImpl<byte[]>) pulsarClient.newConsumer().topic(topicName)
.subscriptionName(subscriptionName).receiverQueueSize(10).subscriptionType(SubscriptionType.Shared)
@@ -1016,6 +1016,7 @@ public class BatchMessageTest extends BrokerTestBase {
producer.close();
consumer1.close();
+ consumer2.close();
}
@Test(dataProvider="testSubTypeAndEnableBatch")
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/DeadLetterPolicy.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/DeadLetterPolicy.java
index f0b75ff57dd..c2a172666b0 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/DeadLetterPolicy.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/DeadLetterPolicy.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.client.api;
+import java.io.Serializable;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
@@ -36,7 +37,8 @@ import org.apache.pulsar.common.classification.InterfaceStability;
@AllArgsConstructor
@InterfaceAudience.Public
@InterfaceStability.Stable
-public class DeadLetterPolicy {
+public class DeadLetterPolicy implements Serializable {
+ private static final long serialVersionUID = 1L;
/**
* Maximum number of times that a message will be redelivered before being sent to the dead letter queue.
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/KeySharedPolicy.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/KeySharedPolicy.java
index f5bc5b846b6..ccaed04d75d 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/KeySharedPolicy.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/KeySharedPolicy.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.client.api;
+import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
@@ -29,7 +30,7 @@ import org.apache.pulsar.common.classification.InterfaceStability;
*/
@InterfaceAudience.Public
@InterfaceStability.Stable
-public abstract class KeySharedPolicy {
+public abstract class KeySharedPolicy implements Serializable {
protected KeySharedMode keySharedMode;
@@ -82,6 +83,7 @@ public abstract class KeySharedPolicy {
* for message, the cursor will rewind.
*/
public static class KeySharedPolicySticky extends KeySharedPolicy {
+ private static final long serialVersionUID = 1L;
protected final List<Range> ranges;
@@ -129,6 +131,7 @@ public abstract class KeySharedPolicy {
* Auto split hash range key shared policy.
*/
public static class KeySharedPolicyAutoSplit extends KeySharedPolicy {
+ private static final long serialVersionUID = 1L;
KeySharedPolicyAutoSplit() {
this.keySharedMode = KeySharedMode.AUTO_SPLIT;
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Range.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Range.java
index 488083f484b..3af3b9c9811 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Range.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Range.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.client.api;
+import java.io.Serializable;
import java.util.Objects;
import org.apache.pulsar.common.classification.InterfaceAudience;
import org.apache.pulsar.common.classification.InterfaceStability;
@@ -27,7 +28,8 @@ import org.apache.pulsar.common.classification.InterfaceStability;
*/
@InterfaceAudience.Public
@InterfaceStability.Stable
-public class Range implements Comparable<Range> {
+public class Range implements Comparable<Range>, Serializable {
+ private static final long serialVersionUID = 1L;
private final int start;
private final int end;
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java
index 88a0aa8ec13..8ef5b9b2528 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java
@@ -359,7 +359,7 @@ public class ConsumerConfigurationData<T> implements Serializable, Cloneable {
+ "When specifying the dead letter policy while not specifying `ackTimeoutMillis`, you can set the"
+ " ack timeout to 30000 millisecond."
)
- private transient DeadLetterPolicy deadLetterPolicy;
+ private DeadLetterPolicy deadLetterPolicy;
private boolean retryEnable = false;
@@ -388,7 +388,7 @@ public class ConsumerConfigurationData<T> implements Serializable, Cloneable {
private boolean resetIncludeHead = false;
@JsonIgnore
- private transient KeySharedPolicy keySharedPolicy;
+ private KeySharedPolicy keySharedPolicy;
private boolean batchIndexAckEnabled = false;
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ReaderConfigurationData.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ReaderConfigurationData.java
index 86707d2aa2f..d1cd4941481 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ReaderConfigurationData.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ReaderConfigurationData.java
@@ -139,7 +139,7 @@ public class ReaderConfigurationData<T> implements Serializable, Cloneable {
)
private boolean resetIncludeHead = false;
- private transient List<Range> keyHashRanges;
+ private List<Range> keyHashRanges;
private boolean poolMessages = false;
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationDataTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationDataTest.java
index f47f83bcbce..291583c3067 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationDataTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationDataTest.java
@@ -19,7 +19,18 @@
package org.apache.pulsar.client.impl.conf;
import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.Collections;
import java.util.regex.Pattern;
+
+import lombok.Cleanup;
+import org.apache.pulsar.client.api.DeadLetterPolicy;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.testng.Assert;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@@ -45,4 +56,43 @@ public class ConsumerConfigurationDataTest {
assertThat(topicConsumerConfigurationData.getPriorityLevel()).isEqualTo(expectedPriority);
}
+
+ @Test
+ public void testSerializable() throws Exception {
+ ConsumerConfigurationData<String> consumerConfigurationData = new ConsumerConfigurationData<>();
+ consumerConfigurationData.setPriorityLevel(1);
+ consumerConfigurationData.setSubscriptionName("my-sub");
+ consumerConfigurationData.setSubscriptionType(SubscriptionType.Shared);
+ consumerConfigurationData.setReceiverQueueSize(100);
+ consumerConfigurationData.setAckTimeoutMillis(1000);
+ consumerConfigurationData.setTopicNames(Collections.singleton("my-topic"));
+
+ DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder()
+ .maxRedeliverCount(10)
+ .retryLetterTopic("retry-topic")
+ .deadLetterTopic("dead-topic")
+ .build();
+ consumerConfigurationData.setDeadLetterPolicy(deadLetterPolicy);
+
+ @Cleanup
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ @Cleanup
+ ObjectOutputStream oos = new ObjectOutputStream(bos);
+ oos.writeObject(consumerConfigurationData);
+ byte[] serialized = bos.toByteArray();
+
+ // Deserialize
+ @Cleanup
+ ByteArrayInputStream bis = new ByteArrayInputStream(serialized);
+ @Cleanup
+ ObjectInputStream ois = new ObjectInputStream(bis);
+ Object object = ois.readObject();
+
+ Assert.assertEquals(object.getClass(), ConsumerConfigurationData.class);
+ Assert.assertEquals(object, consumerConfigurationData);
+
+ DeadLetterPolicy deserialisedDeadLetterPolicy = ((ConsumerConfigurationData<?>) object).getDeadLetterPolicy();
+ Assert.assertNotNull(deserialisedDeadLetterPolicy);
+ Assert.assertEquals(deserialisedDeadLetterPolicy, deadLetterPolicy);
+ }
}
(pulsar) 02/03: [improve][log] Print ZK path if write to ZK fails due to data being too large to persist (#23652)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit f9f5be5daa80711dd8d2e0ccaff5c033c4c6e898
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Mon Dec 30 10:40:37 2024 +0800
[improve][log] Print ZK path if write to ZK fails due to data being too large to persist (#23652)
(cherry picked from commit 5a3a1f169a7f90181bd5c213c8e9f479bc74f0f2)
---
.../org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java | 6 +++---
.../java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java | 8 +++++++-
.../java/org/apache/pulsar/metadata/impl/batching/MetadataOp.java | 2 ++
3 files changed, 12 insertions(+), 4 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
index 404215c084a..655e8d0f2d8 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
@@ -2188,6 +2188,8 @@ public class ManagedCursorImpl implements ManagedCursor {
if (State.NoLedger.equals(STATE_UPDATER.get(this))) {
if (ledger.isNoMessagesAfterPos(mdEntry.newPosition)) {
+ log.error("[{}][{}] Metadata ledger creation failed, try to persist the position in the metadata"
+ + " store.", ledger.getName(), name);
persistPositionToMetaStore(mdEntry, cb);
} else {
cb.operationFailed(new ManagedLedgerException("Switch new cursor ledger failed"));
@@ -2863,9 +2865,7 @@ public class ManagedCursorImpl implements ManagedCursor {
@Override
public void operationFailed(ManagedLedgerException exception) {
- log.error("[{}][{}] Metadata ledger creation failed {}, try to persist the position in the metadata"
- + " store.", ledger.getName(), name, exception);
-
+ log.error("[{}][{}] Metadata ledger creation failed {}", ledger.getName(), name, exception);
synchronized (pendingMarkDeleteOps) {
// At this point we don't have a ledger ready
STATE_UPDATER.set(ManagedCursorImpl.this, State.NoLedger);
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java
index 2e88cb33324..8cfb23faad3 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java
@@ -32,6 +32,7 @@ import java.util.stream.Collectors;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.zookeeper.BoundExponentialBackoffRetryPolicy;
+import org.apache.commons.lang3.tuple.Pair;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.metadata.api.GetResult;
import org.apache.pulsar.metadata.api.MetadataStore;
@@ -200,10 +201,15 @@ public class ZKMetadataStore extends AbstractBatchedMetadataStore
Collectors.groupingBy(MetadataOp::getType, Collectors.summingInt(op -> 1)))
.entrySet().stream().map(e -> e.getValue() + " " + e.getKey().name() + " entries")
.collect(Collectors.joining(", "));
+ List<Pair> opsForLog = ops.stream()
+ .filter(item -> item.size() > 256 * 1024)
+ .map(op -> Pair.of(op.getPath(), op.size()))
+ .collect(Collectors.toList());
Long totalSize = ops.stream().collect(Collectors.summingLong(MetadataOp::size));
log.warn("Connection loss while executing batch operation of {} "
+ "of total data size of {}. "
- + "Retrying individual operations one-by-one.", countsByType, totalSize);
+ + "Retrying individual operations one-by-one. ops whose size > 256KB: {}",
+ countsByType, totalSize, opsForLog);
// Retry with the individual operations
executor.schedule(() -> {
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/batching/MetadataOp.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/batching/MetadataOp.java
index abf60f7b724..06ff425372b 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/batching/MetadataOp.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/batching/MetadataOp.java
@@ -51,4 +51,6 @@ public interface MetadataOp {
default OpPut asPut() {
return (OpPut) this;
}
+
+ String getPath();
}
(pulsar) 03/03: [fix][client] Cannot access message data inside ProducerInterceptor#onSendAcknowledgement (#23791)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c981972c2122a0ac4bbb08a3da0bfe66e0e54efd
Author: Yike Xiao <km...@live.com>
AuthorDate: Tue Dec 31 02:25:09 2024 +0800
[fix][client] Cannot access message data inside ProducerInterceptor#onSendAcknowledgement (#23791)
(cherry picked from commit fdfe00a3c270cc7bf630224957d9fa86958d9e6e)
---
.../apache/pulsar/client/api/InterceptorsTest.java | 44 ++++++++++++++++++++++
.../apache/pulsar/client/impl/ProducerImpl.java | 22 ++++++-----
2 files changed, 56 insertions(+), 10 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/InterceptorsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/InterceptorsTest.java
index 4fac35aa78f..db31f816d6c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/InterceptorsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/InterceptorsTest.java
@@ -21,10 +21,12 @@ package org.apache.pulsar.client.api;
import com.google.common.collect.Sets;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -215,6 +217,48 @@ public class InterceptorsTest extends ProducerConsumerBase {
producer.close();
}
+ @Test
+ public void testProducerInterceptorAccessMessageData() throws PulsarClientException {
+ List<String> messageDataInBeforeSend = Collections.synchronizedList(new ArrayList<>());
+ List<String> messageDataOnSendAcknowledgement = Collections.synchronizedList(new ArrayList<>());
+ ProducerInterceptor<String> interceptor = new ProducerInterceptor<>() {
+ @Override
+ public void close() {
+ }
+
+ @Override
+ public Message<String> beforeSend(Producer<String> producer, Message<String> message) {
+ messageDataInBeforeSend.add(new String(message.getData()));
+ return message;
+ }
+
+ @Override
+ public void onSendAcknowledgement(Producer<String> producer, Message<String> message, MessageId msgId,
+ Throwable exception) {
+ messageDataOnSendAcknowledgement.add(new String(message.getData()));
+ }
+ };
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic("persistent://my-property/my-ns/my-topic")
+ .intercept(interceptor)
+ .create();
+
+ final String messageValue = UUID.randomUUID().toString();
+ try {
+ producer.newMessage().value(messageValue).send();
+ } catch (Exception ignore) {
+ }
+ Assert.assertEquals(messageDataInBeforeSend.size(), 1,
+ "Message data should be available in beforeSend");
+ Assert.assertEquals(messageDataInBeforeSend.get(0), messageValue,
+ "Message data should be available in beforeSend");
+ Assert.assertEquals(messageDataOnSendAcknowledgement.size(), 1,
+ "Message data should be available in onSendAcknowledgement");
+ Assert.assertEquals(messageDataOnSendAcknowledgement.get(0), messageValue,
+ "Message data should be available in onSendAcknowledgement");
+ }
+
@Test
public void testConsumerInterceptorWithErrors() throws PulsarClientException {
ConsumerInterceptor<String> interceptor = new ConsumerInterceptor<String>() {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
index 77894e9e44b..92f8332dfb2 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
@@ -393,17 +393,19 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
if (payload == null) {
log.error("[{}] [{}] Payload is null when calling onSendComplete, which is not expected.",
topic, producerName);
- } else {
- ReferenceCountUtil.safeRelease(payload);
}
- if (e != null) {
- stats.incrementSendFailed();
- onSendAcknowledgement(msg, null, e);
- sendCallback.getFuture().completeExceptionally(e);
- } else {
- stats.incrementNumAcksReceived(latencyNanos);
- onSendAcknowledgement(msg, msg.getMessageId(), null);
- sendCallback.getFuture().complete(msg.getMessageId());
+ try {
+ if (e != null) {
+ stats.incrementSendFailed();
+ onSendAcknowledgement(msg, null, e);
+ sendCallback.getFuture().completeExceptionally(e);
+ } else {
+ stats.incrementNumAcksReceived(latencyNanos);
+ onSendAcknowledgement(msg, msg.getMessageId(), null);
+ sendCallback.getFuture().complete(msg.getMessageId());
+ }
+ } finally {
+ ReferenceCountUtil.safeRelease(payload);
}
}