Skip to content

Fix #199: Do not stop to replicate when producer throws exception #220

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
merlimat merged 4 commits into from
Feb 17, 2017
Merged

Fix #199: Do not stop to replicate when producer throws exception #220

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
c4eeca5
Fix #199: Do not stop to replicate when producer throws exception
Feb 15, 2017
df454ab
Fix log messages
Feb 15, 2017
b8a99ea
testReplicatorProducerClosing shoud be executed at last since it clos…
Feb 17, 2017
c6afe6e
Add unit test for replication resumption on backlog exceeded
Feb 16, 2017
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
16 changes: 8 additions & 8 deletions ...broker/src/main/java/com/yahoo/pulsar/broker/service/persistent/PersistentReplicator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -392,15 +392,15 @@ public void sendComplete(Exception exception) {
if (exception != null) {
log.error("[{}][{} -> {}] Error producing on remote broker", replicator.topicName,
replicator.localCluster, replicator.remoteCluster, exception);
return;
}

if (log.isDebugEnabled()) {
log.debug("[{}][{} -> {}] Message persisted on remote broker", replicator.topicName,
replicator.localCluster, replicator.remoteCluster);
// cursor shoud be rewinded since it was incremented when readMoreEntries
replicator.cursor.rewind();

@saandrews saandrews Feb 17, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

@merlimat Do you see any side effect if we rewind it for every exception. All failed pending messages would reach here and invoke rewind.

@merlimat merlimat Feb 17, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Rewind is a cheap operation, it just resets the readPosition to markDeletePosition + 1

} else {
if (log.isDebugEnabled()) {
log.debug("[{}][{} -> {}] Message persisted on remote broker", replicator.topicName,
replicator.localCluster, replicator.remoteCluster);
}
replicator.cursor.asyncDelete(entry.getPosition(), replicator, entry.getPosition());
}

replicator.cursor.asyncDelete(entry.getPosition(), replicator, entry.getPosition());
entry.release();

int pending = PENDING_MESSAGES_UPDATER.decrementAndGet(replicator);
Expand Down
82 changes: 76 additions & 6 deletions pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/ReplicatorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
package com.yahoo.pulsar.broker.service;

import static org.mockito.Mockito.spy;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
Expand Down Expand Up @@ -56,11 +57,15 @@
import com.yahoo.pulsar.client.api.MessageBuilder;
import com.yahoo.pulsar.client.api.Producer;
import com.yahoo.pulsar.client.api.PulsarClient;
import com.yahoo.pulsar.client.api.PulsarClientException;
import com.yahoo.pulsar.client.impl.PulsarClientImpl;
import com.yahoo.pulsar.client.impl.ProducerImpl;
import com.yahoo.pulsar.common.naming.DestinationName;
import com.yahoo.pulsar.common.naming.NamespaceBundle;
import com.yahoo.pulsar.common.naming.NamespaceName;
import com.yahoo.pulsar.common.policies.data.BacklogQuota;
import com.yahoo.pulsar.common.policies.data.BacklogQuota.RetentionPolicy;
import com.yahoo.pulsar.common.policies.data.PersistentTopicStats;
import com.yahoo.pulsar.common.policies.data.ReplicatorStats;
import com.yahoo.pulsar.common.util.collections.ConcurrentOpenHashMap;

Expand Down Expand Up @@ -165,18 +170,18 @@ public Void call() throws Exception {

// Case 3: TODO: Once automatic cleanup is implemented, add tests case to verify auto removal of clusters
}

@Test
public void testConcurrentReplicator() throws Exception {

log.info("--- Starting ReplicatorTest::testConfigChange ---");
log.info("--- Starting ReplicatorTest::testConcurrentReplicator ---");

final DestinationName dest = DestinationName.get(String.format("persistent://pulsar/global/ns1/topic-%d", 0));
ClientConfiguration conf = new ClientConfiguration();
conf.setStatsInterval(0, TimeUnit.SECONDS);
Producer producer = PulsarClient.create(url1.toString(), conf).createProducer(dest.toString());
producer.close();

PersistentTopic topic = (PersistentTopic) pulsar1.getBrokerService().getTopic(dest.toString()).get();

PulsarClientImpl pulsarClient = spy((PulsarClientImpl) pulsar1.getBrokerService().getReplicationClient("r3"));
Expand Down Expand Up @@ -344,7 +349,7 @@ public Void call() throws Exception {
@Test(enabled = false)
public void testReplicationOverrides() throws Exception {

log.info("--- Starting ReplicatorTest::testReplication ---");
log.info("--- Starting ReplicatorTest::testReplicationOverrides ---");

// This test is to verify that the config change on global namespace is successfully applied in broker during
// runtime.
Expand Down Expand Up @@ -608,7 +613,7 @@ public Void call() throws Exception {
}
}
}

/**
* It verifies that: if it fails while removing replicator-cluster-cursor: it should not restart the replicator and
* it should have cleaned up from the list
Expand Down Expand Up @@ -652,7 +657,7 @@ public void deleteCursorFailed(ManagedLedgerException exception, Object ctx) {
});
}

@Test
@Test(priority = 5)
public void testReplicatorProducerClosing() throws Exception {
log.info("--- Starting ReplicatorTest::testDeleteReplicatorFailure ---");
final String topicName = "persistent://pulsar/global/ns/repltopicbatch";
Expand All @@ -671,6 +676,71 @@ public void testReplicatorProducerClosing() throws Exception {
assertNull(producer);
}

/**
* Issue #199
*
* It verifies that: if the remote cluster reaches backlog quota limit, replicator temporarily stops and once the
* backlog drains it should resume replication.
*
* @throws Exception
*/

@Test(enabled = true, priority = -1)
public void testResumptionAfterBacklogRelaxed() throws Exception {

List<RetentionPolicy> policies = Lists.newArrayList();
policies.add(RetentionPolicy.producer_exception);
policies.add(RetentionPolicy.producer_request_hold);

for (RetentionPolicy policy : policies) {

DestinationName dest = DestinationName.get(String.format("persistent://pulsar/global/ns1/%s", policy));

// Producer on r1
MessageProducer producer1 = new MessageProducer(url1, dest);

// Consumer on r1
MessageConsumer consumer1 = new MessageConsumer(url1, dest);

// Consumer on r2
MessageConsumer consumer2 = new MessageConsumer(url2, dest);

// Replicator for r1 -> r2
PersistentTopic topic = (PersistentTopic) pulsar1.getBrokerService().getTopicReference(dest.toString());
PersistentReplicator replicator = topic.getPersistentReplicator("r2");

// Restrict backlog quota limit to 1
admin1.namespaces().setBacklogQuota("pulsar/global/ns1", new BacklogQuota(1, policy));

// Produce a message to r1, then it will be replicated to r2 and fulfill the backlog.
producer1.produce(1);
consumer1.receive(1);
Thread.sleep((TIME_TO_CHECK_BACKLOG_QUOTA + 1) * 1000);

// Produce 9 messages to r1, then it will be pended because of the backlog limit excess
producer1.produce(9);
consumer1.receive(9);
Thread.sleep(1000L);
assertEquals(replicator.getStats().replicationBacklog, 9);

// Relax backlog quota limit to 1G
admin1.namespaces().setBacklogQuota("pulsar/global/ns1", new BacklogQuota(1024 * 1024 * 1024, policy));
Thread.sleep((TIME_TO_CHECK_BACKLOG_QUOTA + 1) * 1000);

// The messages should be replicated to r2
assertEquals(replicator.getStats().replicationBacklog, 0);
consumer2.receive(1);
consumer2.receive(9);
if (!consumer2.drained()) {
throw new Exception("consumer2 - unexpected message in queue");
}

producer1.close();
consumer1.close();
consumer2.close();
}
}

private static final Logger log = LoggerFactory.getLogger(ReplicatorTest.class);

}
4 changes: 4 additions & 0 deletions pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/ReplicatorTestBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,8 @@ public class ReplicatorTestBase {

ExecutorService executor = new ThreadPoolExecutor(5, 20, 30, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());

static final int TIME_TO_CHECK_BACKLOG_QUOTA = 5;

// Default frequency
public int getBrokerServicePurgeInactiveFrequency() {
return 60;
Expand Down Expand Up @@ -111,6 +113,7 @@ void setup() throws Exception {
config1.setBrokerServicePurgeInactiveFrequencyInSeconds(
inSec(getBrokerServicePurgeInactiveFrequency(), TimeUnit.SECONDS));
config1.setBrokerServicePort(PortManager.nextFreePort());
config1.setBacklogQuotaCheckIntervalInSeconds(TIME_TO_CHECK_BACKLOG_QUOTA);
pulsar1 = new PulsarService(config1);
pulsar1.start();
ns1 = pulsar1.getBrokerService();
Expand All @@ -135,6 +138,7 @@ void setup() throws Exception {
config2.setBrokerServicePurgeInactiveFrequencyInSeconds(
inSec(getBrokerServicePurgeInactiveFrequency(), TimeUnit.SECONDS));
config2.setBrokerServicePort(PortManager.nextFreePort());
config2.setBacklogQuotaCheckIntervalInSeconds(TIME_TO_CHECK_BACKLOG_QUOTA);
pulsar2 = new PulsarService(config2);
pulsar2.start();
ns2 = pulsar2.getBrokerService();
Expand Down