Skip to content

[fix] Release EntryImpl while reading exchange topic #1256

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
codelipenghui merged 6 commits into from
Jun 6, 2024
Merged

[fix] Release EntryImpl while reading exchange topic #1256

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
655d928
Release EntryImpl while reading exchange data
gaoran10 Jun 5, 2024
c94f774
fix
gaoran10 Jun 5, 2024
9dd5ef8
fix
gaoran10 Jun 5, 2024
0b7f8fa
fix
gaoran10 Jun 5, 2024
ae3099d
fix
gaoran10 Jun 5, 2024
b4b751c
remove useless log
gaoran10 Jun 6, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
113 changes: 91 additions & 22 deletions amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/AmqpExchangeReplicator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,27 +15,30 @@

import static org.apache.pulsar.broker.service.persistent.PersistentTopic.MESSAGE_RATE_BACKOFF_MS;

import io.netty.buffer.ByteBuf;
import io.streamnative.pulsar.handlers.amqp.impl.PersistentExchange;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.client.impl.Backoff;
import org.apache.pulsar.client.impl.MessageImpl;
import org.apache.pulsar.common.api.proto.CommandSubscribe;
import org.apache.pulsar.common.api.proto.KeyValue;
import org.apache.pulsar.common.util.Backoff;

/**
* Amqp exchange replicator, read entries from BookKeeper and process entries.
Expand Down Expand Up @@ -64,11 +67,13 @@ protected enum State {
Stopped, Starting, Started, Stopping
}

private static final int defaultReadMaxSizeBytes = 5 * 1024 * 1024;
private int routeQueueSize = 200;
private volatile int pendingQueueSize = 0;
private static final AtomicIntegerFieldUpdater<AmqpExchangeReplicator> PENDING_SIZE_UPDATER =
AtomicIntegerFieldUpdater.newUpdater(AmqpExchangeReplicator.class, "pendingQueueSize");
private int readBatchSize;
private final int readMaxSizeBytes;
private final int readMaxBatchSize;

private static final int FALSE = 0;
private static final int TRUE = 1;
Expand All @@ -89,6 +94,10 @@ protected AmqpExchangeReplicator(PersistentExchange persistentExchange, Executor
this.scheduledExecutorService = topic.getBrokerService().executor();
this.initMaxRouteQueueSize(routeQueueSize);
this.routeExecutor = routeExecutor;
this.readMaxBatchSize = Math.min(this.routeQueueSize,
topic.getBrokerService().getPulsar().getConfig().getDispatcherMaxReadBatchSize());
this.readBatchSize = this.readMaxBatchSize;
this.readMaxSizeBytes = topic.getBrokerService().getPulsar().getConfig().getDispatcherMaxReadSizeBytes();
STATE_UPDATER.set(this, AmqpExchangeReplicator.State.Stopped);
this.name = "[AMQP Replicator for " + topic.getName() + " ]";
}
Expand Down Expand Up @@ -175,11 +184,16 @@ private void readMoreEntries() {
}
int availablePermits = getAvailablePermits();
if (availablePermits > 0) {
int messagesToRead = Math.min(availablePermits, readBatchSize);
// avoid messageToRead is 0
messagesToRead = Math.max(messagesToRead, 1);

if (HAVE_PENDING_READ_UPDATER.compareAndSet(this, FALSE, TRUE)) {
log.info("{} Schedule read of {} messages.", name, messagesToRead);
if (log.isDebugEnabled()) {
log.debug("{} Schedule read of {} messages.", name, availablePermits);
log.debug("{} Schedule read of {} messages.", name, messagesToRead);
}
cursor.asyncReadEntriesOrWait(availablePermits, defaultReadMaxSizeBytes, this, null, null);
cursor.asyncReadEntriesOrWait(messagesToRead, readMaxSizeBytes, this, null, null);
} else {
if (log.isDebugEnabled()) {
log.debug("{} Not schedule read due to pending read. Messages to read {}.",
Expand All @@ -201,6 +215,23 @@ private int getAvailablePermits() {
}
return 0;
}

if (topic.getDispatchRateLimiter().isPresent()
&& topic.getDispatchRateLimiter().get().isDispatchRateLimitingEnabled()) {
long availableOnByte = topic.getDispatchRateLimiter().get().getAvailableDispatchRateLimitOnByte();
long availableOnMsg = topic.getDispatchRateLimiter().get().getAvailableDispatchRateLimitOnMsg();
if (availableOnByte == 0 || availableOnMsg == 0) {
if (log.isDebugEnabled()) {
log.debug("{} Dispatch rate limit is reached, availableOnByte: {}, availableOnMsg: {}.",
name, availableOnByte, availableOnMsg);
}
return -1;
}
if (availableOnMsg > 0) {
availablePermits = Math.min(availablePermits, (int) availableOnMsg);
}
}

return availablePermits;
}

Expand All @@ -210,46 +241,83 @@ public void readEntriesComplete(List<Entry> list, Object o) {
log.debug("{} Read entries complete. Entries size: {}", name, list.size());
}
HAVE_PENDING_READ_UPDATER.set(this, FALSE);
if (list == null || list.isEmpty()) {
if (CollectionUtils.isEmpty(list)) {
long delay = readFailureBackoff.next();
log.warn("{} The read entry list is empty, will retry in {} ms. ReadPosition: {}, LAC: {}.",
name, delay, cursor.getReadPosition(), topic.getManagedLedger().getLastConfirmedEntry());
scheduledExecutorService.schedule(this::readMoreEntries, delay, TimeUnit.MILLISECONDS);
return;
}

if (readBatchSize < readMaxBatchSize) {
int newReadBatchSize = Math.min(readBatchSize * 2, readMaxBatchSize);
if (log.isDebugEnabled()) {
log.debug("[{}] Increasing read batch size from {} to {}", name, readBatchSize,
newReadBatchSize);
}

readBatchSize = newReadBatchSize;
}

readFailureBackoff.reduceToHalf();
List<Pair<PositionImpl, ByteBuf>> bufList = new ArrayList<>(list.size());
routeExecutor.execute(() -> this.handleEntries(list));
}

private void handleEntries(List<Entry> list) {
PENDING_SIZE_UPDATER.addAndGet(this, list.size());

List<Pair<Position, Map<String, Object>>> propsList = new ArrayList<>();
boolean encounterError = false;
for (Entry entry : list) {
bufList.add(
Pair.of(PositionImpl.get(entry.getLedgerId(), entry.getEntryId()), entry.getDataBuffer()));
if (encounterError) {
entry.release();
continue;
}

Map<String, Object> props;
try {
MessageImpl<byte[]> message = MessageImpl.deserialize(entry.getDataBuffer());
props = message.getMessageBuilder().getPropertiesList().stream()
.collect(Collectors.toMap(KeyValue::getKey, KeyValue::getValue));
} catch (Exception e) {
log.error("Failed to deserialize entry dataBuffer for topic: {}, rewind cursor.", name, e);
encounterError = true;
propsList.clear();
continue;
}

propsList.add(Pair.of(entry.getPosition(), props));
topic.getDispatchRateLimiter().ifPresent(
limiter -> limiter.consumeDispatchQuota(1, entry.getLength()));
entry.release();

@gaoran10 gaoran10 Jun 5, 2024

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Release the entry after collecting route information.

}
if (encounterError) {
cursor.rewind();
PENDING_SIZE_UPDATER.set(this, 0);
this.readMoreEntries();
return;
}
routeExecutor.execute(() -> this.readComplete(bufList));
}

private void readComplete(List<Pair<PositionImpl, ByteBuf>> list) {
for (Pair<PositionImpl, ByteBuf> entry : list) {
PENDING_SIZE_UPDATER.incrementAndGet(this);
readProcess(entry.getRight(), entry.getLeft()).whenCompleteAsync((ignored, exception) -> {
for (var posAndProps : propsList) {
final Position position = posAndProps.getLeft();
routeIndex(posAndProps.getRight(), position).whenCompleteAsync((ignored, exception) -> {
if (exception != null) {
log.error("{} Error producing messages", name, exception);
this.cursor.rewind();
} else {
if (log.isDebugEnabled()) {
log.debug("{} Route message successfully.", name);
}
AmqpExchangeReplicator.this.cursor
.asyncDelete(entry.getLeft(), this, entry.getLeft());
AmqpExchangeReplicator.this.cursor.asyncDelete(position, this, position);
}
if (PENDING_SIZE_UPDATER.decrementAndGet(this) < routeQueueSize * 0.5
&& HAVE_PENDING_READ_UPDATER.get(this) == FALSE) {
if (PENDING_SIZE_UPDATER.decrementAndGet(this) <= 0) {
this.readMoreEntries();
}
}, routeExecutor);
entry.getRight().release();

@gaoran10 gaoran10 Jun 5, 2024

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is the key point, we should release the entry directly.

}
}

public abstract CompletableFuture<Void> readProcess(ByteBuf data, Position position);
public abstract CompletableFuture<Void> routeIndex(Map<String, Object> props, Position position);

@Override
public void readEntriesFailed(ManagedLedgerException exception, Object o) {
Expand All @@ -266,6 +334,7 @@ public void readEntriesFailed(ManagedLedgerException exception, Object o) {
log.debug("{} Read entries from bookie failed, retrying in {} s", name, waitTimeMs / 1000, exception);
}
HAVE_PENDING_READ_UPDATER.set(this, FALSE);
readBatchSize = topic.getBrokerService().pulsar().getConfiguration().getDispatcherMinReadBatchSize();
scheduledExecutorService.schedule(this::readMoreEntries, waitTimeMs, TimeUnit.MILLISECONDS);
}

Expand Down
4 changes: 2 additions & 2 deletions amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/AmqpPulsarConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,9 +23,9 @@
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.impl.Backoff;
import org.apache.pulsar.client.impl.BackoffBuilder;
import org.apache.pulsar.client.impl.MessageIdImpl;
import org.apache.pulsar.common.util.Backoff;
import org.apache.pulsar.common.util.BackoffBuilder;
import org.apache.qpid.server.protocol.v0_8.AMQShortString;

/**
Expand Down
15 changes: 1 addition & 14 deletions amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/impl/PersistentExchange.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.google.common.collect.Sets;
import io.netty.buffer.ByteBuf;
import io.streamnative.pulsar.handlers.amqp.AbstractAmqpExchange;
import io.streamnative.pulsar.handlers.amqp.AmqpEntryWriter;
import io.streamnative.pulsar.handlers.amqp.AmqpExchangeReplicator;
Expand All @@ -35,7 +34,6 @@
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.EqualsAndHashCode;
Expand All @@ -51,9 +49,7 @@
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.impl.MessageImpl;
import org.apache.pulsar.common.api.proto.CommandSubscribe;
import org.apache.pulsar.common.api.proto.KeyValue;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicDomain;
import org.apache.pulsar.common.naming.TopicName;
Expand Down Expand Up @@ -126,16 +122,7 @@ public PersistentExchange(String exchangeName, Type type, PersistentTopic persis
if (messageReplicator == null) {
messageReplicator = new AmqpExchangeReplicator(this, routeExecutor, routeQueueSize) {
@Override
public CompletableFuture<Void> readProcess(ByteBuf data, Position position) {
Map<String, Object> props;
try {
MessageImpl<byte[]> message = MessageImpl.deserialize(data);
props = message.getMessageBuilder().getPropertiesList().stream()
.collect(Collectors.toMap(KeyValue::getKey, KeyValue::getValue));
} catch (Exception e) {
log.error("Failed to deserialize entry dataBuffer. exchangeName: {}", exchangeName, e);
return FutureUtil.failedFuture(e);
}
public CompletableFuture<Void> routeIndex(Map<String, Object> props, Position position) {

List<CompletableFuture<Void>> routeFutureList = new ArrayList<>();
if (exchangeType == Type.Direct) {
Expand Down
1 change: 1 addition & 0 deletions amqp-impl/src/test/java/io/streamnative/pulsar/handlers/amqp/test/AmqpProtocolTestBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,6 +148,7 @@ public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
}
});
when(pulsarService.getConfiguration()).thenReturn(serviceConfiguration);
when(pulsarService.getConfig()).thenReturn(serviceConfiguration);
when(pulsarService.getOrderedExecutor()).thenReturn(
OrderedExecutor.newBuilder().numThreads(8).name("pulsar-ordered").build());
when(serviceConfiguration.getNumIOThreads()).thenReturn(2 * Runtime.getRuntime().availableProcessors());
Expand Down
3 changes: 3 additions & 0 deletions tests/src/test/java/io/streamnative/pulsar/handlers/amqp/AmqpProtocolHandlerTestBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,8 @@ public abstract class AmqpProtocolHandlerTestBase {
@Getter
private List<Integer> aopAdminPortList = new ArrayList<>();

private final Integer inFlightSizeInMB = 5;
Comment thread
codelipenghui marked this conversation as resolved.

public AmqpProtocolHandlerTestBase() {
resetConfig();
}
Expand All @@ -128,6 +130,7 @@ protected void resetConfig() {
amqpConfig.setBrokerShutdownTimeoutMs(0L);
amqpConfig.setDefaultNumPartitions(1);
amqpConfig.setTransactionCoordinatorEnabled(true);
amqpConfig.setManagedLedgerMaxReadsInFlightSizeInMB(inFlightSizeInMB);

// set protocol related config
URL testHandlerUrl = this.getClass().getClassLoader().getResource("test-protocol-handler.nar");
Expand Down
38 changes: 38 additions & 0 deletions tests/src/test/java/io/streamnative/pulsar/handlers/amqp/rabbitmq/RabbitMQMessagingTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,8 @@
*/
package io.streamnative.pulsar.handlers.amqp.rabbitmq;

import static org.junit.Assert.assertTrue;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
Expand All @@ -33,9 +35,11 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomUtils;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClient;
Expand Down Expand Up @@ -512,4 +516,38 @@ public void handleDelivery(String consumerTag,
channel.close();
conn.close();
}

@Test(timeOut = 15_000)
public void testConsumeMoreThanInFlightSize() throws Exception {
Connection conn = getConnection("vhost1", true);
Channel channel = conn.createChannel();

String qu = randQuName();
channel.queueDeclare(qu, true, false, false, null);

long totalContentSize = conf.getManagedLedgerMaxReadsInFlightSizeInMB() * 1024 * 1024 * 2;
AtomicLong receiveContentSize = new AtomicLong(0);

CountDownLatch latch = new CountDownLatch(1);
channel.basicConsume(qu, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
byte[] body) throws IOException {
if (receiveContentSize.addAndGet(body.length) >= totalContentSize) {
latch.countDown();
}
}
});

AtomicLong sendContentSize = new AtomicLong(0);
byte[] content = RandomUtils.nextBytes(1024 * 100);
do {
channel.basicPublish("", qu, null, content);
} while (sendContentSize.addAndGet(content.length) < totalContentSize);

assertTrue(latch.await(10, TimeUnit.SECONDS));
channel.close();
conn.close();
}

}