Skip to content
Merged
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -83,7 +83,7 @@ private boolean topicHasSchema(String topicName) {
@Test(groups = "broker")
public void testGCWillDeleteSchema() throws Exception {
// 1. Simple successful GC
String topicName = "non-persistent://prop/ns-abc/topic-1";
final String topicName = "non-persistent://prop/ns-abc/topic-1";
Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
producer.close();

Expand All @@ -100,56 +100,57 @@ public void testGCWillDeleteSchema() throws Exception {
assertTrue(topicHasSchema(topicName));
runGC();

topic = getTopic(topicName);
assertFalse(topic.isPresent());
Awaitility.await().untilAsserted(() -> {
assertFalse(getTopic(topicName).isPresent());
});
assertFalse(topicHasSchema(topicName));

// 1a. Topic that add/removes subscription can be GC'd
topicName = "non-persistent://prop/ns-abc/topic-1a";
final String topicName2 = "non-persistent://prop/ns-abc/topic-1a";
String subName = "sub1";
Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName(subName).subscribe();
topic = getTopic(topicName);
Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName2).subscriptionName(subName).subscribe();
topic = getTopic(topicName2);
assertTrue(topic.isPresent());
topic.get().addSchema(schemaData).join();
assertTrue(topicHasSchema(topicName));
assertTrue(topicHasSchema(topicName2));

admin.topics().deleteSubscription(topicName, subName);
admin.topics().deleteSubscription(topicName2, subName);
consumer.close();

runGC();
topic = getTopic(topicName);
assertFalse(topic.isPresent());
assertFalse(topicHasSchema(topicName));
Awaitility.await().untilAsserted(() -> {
assertFalse(getTopic(topicName2).isPresent());
});
assertFalse(topicHasSchema(topicName2));

// 2. Topic is not GCed with live connection
topicName = "non-persistent://prop/ns-abc/topic-2";
final String topicName3 = "non-persistent://prop/ns-abc/topic-2";
subName = "sub1";
consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName(subName).subscribe();
topic = getTopic(topicName);
consumer = pulsarClient.newConsumer().topic(topicName3).subscriptionName(subName).subscribe();
topic = getTopic(topicName3);
assertTrue(topic.isPresent());
topic.get().addSchema(schemaData).join();
assertTrue(topicHasSchema(topicName));
assertTrue(topicHasSchema(topicName3));

runGC();
topic = getTopic(topicName);
assertTrue(topic.isPresent());
assertTrue(topicHasSchema(topicName));
assertTrue(getTopic(topicName3).isPresent());
assertTrue(topicHasSchema(topicName3));

// 3. Topic with subscription is not GCed even with no connections
consumer.close();

runGC();
topic = getTopic(topicName);
assertTrue(topic.isPresent());
assertTrue(topicHasSchema(topicName));
assertTrue(getTopic(topicName3).isPresent());
assertTrue(topicHasSchema(topicName3));

// 4. Topic can be GCed after unsubscribe
admin.topics().deleteSubscription(topicName, subName);
admin.topics().deleteSubscription(topicName3, subName);

runGC();
topic = getTopic(topicName);
assertFalse(topic.isPresent());
assertFalse(topicHasSchema(topicName));
Awaitility.await().untilAsserted(() -> {
assertFalse(getTopic(topicName3).isPresent());
});
assertFalse(topicHasSchema(topicName3));
}

@Test(groups = "broker")
Expand Down