Skip to content

Commit

Permalink
Fix blocking check backlog quota in Producer Creation
Browse files Browse the repository at this point in the history
  • Loading branch information
Jiang Haiting committed Nov 18, 2021
1 parent 7a34ceb commit 9fad6cc
Show file tree
Hide file tree
Showing 5 changed files with 104 additions and 74 deletions.
12 changes: 12 additions & 0 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerServiceException.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,11 @@
*/
package org.apache.pulsar.broker.service;

import lombok.Getter;
import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException;
import org.apache.pulsar.broker.service.schema.exceptions.InvalidSchemaDataException;
import org.apache.pulsar.common.api.proto.ServerError;
import org.apache.pulsar.common.policies.data.BacklogQuota;
import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
import org.apache.pulsar.transaction.coordinator.exceptions.CoordinatorException;

Expand Down Expand Up @@ -191,6 +193,16 @@ public TopicPoliciesCacheNotInitException() {
}
}

public static class TopicBacklogQuotaExceededException extends BrokerServiceException {
@Getter
private final BacklogQuota.RetentionPolicy retentionPolicy;

public TopicBacklogQuotaExceededException(BacklogQuota.RetentionPolicy retentionPolicy) {
super("Cannot create producer on topic with backlog quota exceeded");
this.retentionPolicy = retentionPolicy;
}
}

public static org.apache.pulsar.common.api.proto.ServerError getClientErrorCode(Throwable t) {
return getClientErrorCode(t, true);
}
Expand Down
106 changes: 56 additions & 50 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,7 @@
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.BacklogQuota;
import org.apache.pulsar.common.policies.data.BacklogQuota.BacklogQuotaType;
import org.apache.pulsar.common.policies.data.NamespaceOperation;
import org.apache.pulsar.common.policies.data.TopicOperation;
import org.apache.pulsar.common.policies.data.stats.ConsumerStatsImpl;
Expand Down Expand Up @@ -1189,65 +1190,70 @@ protected void handleProducer(final CommandProducer cmdProducer) {
service.getOrCreateTopic(topicName.toString()).thenAccept((Topic topic) -> {
// Before creating producer, check if backlog quota exceeded
// on topic for size based limit and time based limit
for (BacklogQuota.BacklogQuotaType backlogQuotaType : BacklogQuota.BacklogQuotaType.values()) {
if (topic.isBacklogQuotaExceeded(producerName, backlogQuotaType)) {
IllegalStateException illegalStateException = new IllegalStateException(
"Cannot create producer on topic with backlog quota exceeded");
BacklogQuota.RetentionPolicy retentionPolicy = topic
.getBacklogQuota(backlogQuotaType).getPolicy();
if (retentionPolicy == BacklogQuota.RetentionPolicy.producer_request_hold) {
commandSender.sendErrorResponse(requestId,
ServerError.ProducerBlockedQuotaExceededError,
illegalStateException.getMessage());
} else if (retentionPolicy == BacklogQuota.RetentionPolicy.producer_exception) {
commandSender.sendErrorResponse(requestId,
ServerError.ProducerBlockedQuotaExceededException,
illegalStateException.getMessage());
}
producerFuture.completeExceptionally(illegalStateException);
producers.remove(producerId, producerFuture);
return;
CompletableFuture<Void> backlogQuotaCheckFuture = CompletableFuture.allOf(
topic.checkBacklogQuotaExceeded(producerName, BacklogQuotaType.destination_storage),
topic.checkBacklogQuotaExceeded(producerName, BacklogQuotaType.message_age));
backlogQuotaCheckFuture.exceptionally(throwable -> {
//throwable should be CompletionException holding TopicBacklogQuotaExceededException
BrokerServiceException.TopicBacklogQuotaExceededException exception =
(BrokerServiceException.TopicBacklogQuotaExceededException) throwable.getCause();
IllegalStateException illegalStateException = new IllegalStateException(exception);
BacklogQuota.RetentionPolicy retentionPolicy = exception.getRetentionPolicy();
if (retentionPolicy == BacklogQuota.RetentionPolicy.producer_request_hold) {
commandSender.sendErrorResponse(requestId,
ServerError.ProducerBlockedQuotaExceededError,
illegalStateException.getMessage());
} else if (retentionPolicy == BacklogQuota.RetentionPolicy.producer_exception) {
commandSender.sendErrorResponse(requestId,
ServerError.ProducerBlockedQuotaExceededException,
illegalStateException.getMessage());
}
}
// Check whether the producer will publish encrypted messages or not
if ((topic.isEncryptionRequired() || encryptionRequireOnProducer) && !isEncrypted) {
String msg = String.format("Encryption is required in %s", topicName);
log.warn("[{}] {}", remoteAddress, msg);
commandSender.sendErrorResponse(requestId, ServerError.MetadataError, msg);
producerFuture.completeExceptionally(illegalStateException);
producers.remove(producerId, producerFuture);
return;
}
return null;
});

disableTcpNoDelayIfNeeded(topicName.toString(), producerName);
backlogQuotaCheckFuture.thenRun(() -> {
// Check whether the producer will publish encrypted messages or not
if ((topic.isEncryptionRequired() || encryptionRequireOnProducer) && !isEncrypted) {
String msg = String.format("Encryption is required in %s", topicName);
log.warn("[{}] {}", remoteAddress, msg);
commandSender.sendErrorResponse(requestId, ServerError.MetadataError, msg);
producers.remove(producerId, producerFuture);
return;
}

CompletableFuture<SchemaVersion> schemaVersionFuture = tryAddSchema(topic, schema);
disableTcpNoDelayIfNeeded(topicName.toString(), producerName);

schemaVersionFuture.exceptionally(exception -> {
String message = exception.getMessage();
if (exception.getCause() != null) {
message += (" caused by " + exception.getCause());
}
commandSender.sendErrorResponse(requestId,
BrokerServiceException.getClientErrorCode(exception),
message);
producers.remove(producerId, producerFuture);
return null;
});
CompletableFuture<SchemaVersion> schemaVersionFuture = tryAddSchema(topic, schema);

schemaVersionFuture.thenAccept(schemaVersion -> {
topic.checkIfTransactionBufferRecoverCompletely(isTxnEnabled).thenAccept(future -> {
buildProducerAndAddTopic(topic, producerId, producerName, requestId, isEncrypted,
metadata, schemaVersion, epoch, userProvidedProducerName, topicName,
producerAccessMode, topicEpoch, producerFuture);
}).exceptionally(exception -> {
Throwable cause = exception.getCause();
log.error("producerId {}, requestId {} : TransactionBuffer recover failed",
producerId, requestId, exception);
schemaVersionFuture.exceptionally(exception -> {
String message = exception.getMessage();
if (exception.getCause() != null) {
message += (" caused by " + exception.getCause());
}
commandSender.sendErrorResponse(requestId,
ServiceUnitNotReadyException.getClientErrorCode(cause),
cause.getMessage());
BrokerServiceException.getClientErrorCode(exception),
message);
producers.remove(producerId, producerFuture);
return null;
});

schemaVersionFuture.thenAccept(schemaVersion -> {
topic.checkIfTransactionBufferRecoverCompletely(isTxnEnabled).thenAccept(future -> {
buildProducerAndAddTopic(topic, producerId, producerName, requestId, isEncrypted,
metadata, schemaVersion, epoch, userProvidedProducerName, topicName,
producerAccessMode, topicEpoch, producerFuture);
}).exceptionally(exception -> {
Throwable cause = exception.getCause();
log.error("producerId {}, requestId {} : TransactionBuffer recover failed",
producerId, requestId, exception);
commandSender.sendErrorResponse(requestId,
ServiceUnitNotReadyException.getClientErrorCode(cause),
cause.getMessage());
return null;
});
});
});
}).exceptionally(exception -> {
Throwable cause = exception.getCause();
Expand Down
2 changes: 1 addition & 1 deletion pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -194,7 +194,7 @@ CompletableFuture<Subscription> createSubscription(String subscriptionName, Init

CompletableFuture<Void> onPoliciesUpdate(Policies data);

boolean isBacklogQuotaExceeded(String producerName, BacklogQuota.BacklogQuotaType backlogQuotaType);
CompletableFuture<Void> checkBacklogQuotaExceeded(String producerName, BacklogQuota.BacklogQuotaType backlogQuotaType);

boolean isEncryptionRequired();

Expand Down
4 changes: 2 additions & 2 deletions ...oker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -995,9 +995,9 @@ public BacklogQuota getBacklogQuota(BacklogQuota.BacklogQuotaType backlogQuotaTy
* @return quota exceeded status for blocking producer creation
*/
@Override
public boolean isBacklogQuotaExceeded(String producerName, BacklogQuota.BacklogQuotaType backlogQuotaType) {
public CompletableFuture<Void> checkBacklogQuotaExceeded(String producerName, BacklogQuota.BacklogQuotaType backlogQuotaType) {
// No-op
return false;
return CompletableFuture.completedFuture(null);
}

@Override
Expand Down
54 changes: 33 additions & 21 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,7 @@
import org.apache.pulsar.broker.service.BrokerServiceException.ServerMetadataException;
import org.apache.pulsar.broker.service.BrokerServiceException.SubscriptionBusyException;
import org.apache.pulsar.broker.service.BrokerServiceException.SubscriptionNotFoundException;
import org.apache.pulsar.broker.service.BrokerServiceException.TopicBacklogQuotaExceededException;
import org.apache.pulsar.broker.service.BrokerServiceException.TopicBusyException;
import org.apache.pulsar.broker.service.BrokerServiceException.TopicClosedException;
import org.apache.pulsar.broker.service.BrokerServiceException.TopicFencedException;
Expand Down Expand Up @@ -126,6 +127,7 @@
import org.apache.pulsar.common.events.EventsTopicNames;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.BacklogQuota;
import org.apache.pulsar.common.policies.data.BacklogQuota.BacklogQuotaType;
import org.apache.pulsar.common.policies.data.InactiveTopicDeleteMode;
import org.apache.pulsar.common.policies.data.ManagedLedgerInternalStats.CursorStats;
import org.apache.pulsar.common.policies.data.ManagedLedgerInternalStats.LedgerInfo;
Expand Down Expand Up @@ -2463,7 +2465,7 @@ public CompletableFuture<Void> onPoliciesUpdate(Policies data) {
* @return Backlog quota for topic
*/
@Override
public BacklogQuota getBacklogQuota(BacklogQuota.BacklogQuotaType backlogQuotaType) {
public BacklogQuota getBacklogQuota(BacklogQuotaType backlogQuotaType) {
TopicName topicName = TopicName.get(this.getName());
return brokerService.getBacklogQuotaManager().getBacklogQuota(topicName, backlogQuotaType);
}
Expand All @@ -2473,24 +2475,32 @@ public BacklogQuota getBacklogQuota(BacklogQuota.BacklogQuotaType backlogQuotaTy
* @return quota exceeded status for blocking producer creation
*/
@Override
public boolean isBacklogQuotaExceeded(String producerName, BacklogQuota.BacklogQuotaType backlogQuotaType) {
public CompletableFuture<Void> checkBacklogQuotaExceeded(String producerName, BacklogQuotaType backlogQuotaType) {
BacklogQuota backlogQuota = getBacklogQuota(backlogQuotaType);

if (backlogQuota != null) {
BacklogQuota.RetentionPolicy retentionPolicy = backlogQuota.getPolicy();

if ((retentionPolicy == BacklogQuota.RetentionPolicy.producer_request_hold
|| retentionPolicy == BacklogQuota.RetentionPolicy.producer_exception)) {
if (backlogQuotaType == BacklogQuota.BacklogQuotaType.destination_storage && isSizeBacklogExceeded()
|| backlogQuotaType == BacklogQuota.BacklogQuotaType.message_age && isTimeBacklogExceeded()){
if (backlogQuotaType == BacklogQuotaType.destination_storage && isSizeBacklogExceeded()){
log.info("[{}] Backlog quota exceeded. Cannot create producer [{}]", this.getName(), producerName);
return true;
return FutureUtil.failedFuture(new TopicBacklogQuotaExceededException(retentionPolicy));
}
if (backlogQuotaType == BacklogQuotaType.message_age) {
return checkTimeBacklogExceeded().thenCompose(isExceeded->{
if (isExceeded) {
log.info("[{}] Backlog quota exceeded. Cannot create producer [{}]", this.getName(),
producerName);
return FutureUtil.failedFuture(new TopicBacklogQuotaExceededException(retentionPolicy));
} else {
return CompletableFuture.completedFuture(null);
}
});
}
} else {
return false;
return CompletableFuture.completedFuture(null);
}
}
return false;
return CompletableFuture.completedFuture(null);
}

/**
Expand Down Expand Up @@ -2519,17 +2529,25 @@ public boolean isSizeBacklogExceeded() {
* @return determine if backlog quota enforcement needs to be done for topic based on time limit
*/
public boolean isTimeBacklogExceeded() {
try {
return checkTimeBacklogExceeded().get();
} catch (Throwable e) {
log.error("[{}] checkTimeBacklogExceeded failed.", topic, e);
return false;
}
}
public CompletableFuture<Boolean> checkTimeBacklogExceeded() {
TopicName topicName = TopicName.get(getName());
CompletableFuture<Boolean> future = new CompletableFuture<>();
int backlogQuotaLimitInSecond = brokerService.getBacklogQuotaManager().getBacklogQuotaLimitInTime(topicName);

// If backlog quota by time is not set and we have no durable cursor.
if (backlogQuotaLimitInSecond <= 0
|| ((ManagedCursorContainer) ledger.getCursors()).getSlowestReaderPosition() == null) {
return false;
return CompletableFuture.completedFuture(true);
}

if (brokerService.pulsar().getConfiguration().isPreciseTimeBasedBacklogQuotaCheck()) {
CompletableFuture<Boolean> future = new CompletableFuture<>();
// Check if first unconsumed message(first message after mark delete position)
// for slowest cursor's has expired.
PositionImpl position = ((ManagedLedgerImpl) ledger).getNextValidPosition(((ManagedCursorContainer)
Expand Down Expand Up @@ -2562,13 +2580,7 @@ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
future.complete(false);
}
}, null);

try {
return future.get();
} catch (Exception e) {
log.error("[{}][{}] Error reading entry for precise time based backlog check", topicName, e);
return false;
}
return future;
} else {
Long ledgerId = ((ManagedCursorContainer) ledger.getCursors()).getSlowestReaderPosition().getLedgerId();
try {
Expand All @@ -2582,13 +2594,13 @@ public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ "slowest cursor currently on {}", backlogQuotaLimitInSecond * 1000,
((ManagedLedgerImpl) ledger).getClock().millis() - ledgerInfo.getTimestamp());
}
return true;
return CompletableFuture.completedFuture(true);
} else {
return false;
return CompletableFuture.completedFuture(false);
}
} catch (Exception e) {
log.error("[{}][{}] Error reading entry for precise time based backlog check", topicName, e);
return false;
return CompletableFuture.completedFuture(false);
}
}
}
Expand Down

0 comments on commit 9fad6cc

Please sign in to comment.