Merge pull request #1593 from rabbitmq/mk-clear-recorded-bindings-with-deleted-source

acogoluegnes · web-flow · commit e8c193ae99eb · 2025-05-06T07:36:52.000Z
AutorecoveringConnection: when an exchange is deleted, clean up the bindings where it is the source [in the recorded topology]
diff --git a/src/main/java/com/rabbitmq/client/impl/recovery/AutorecoveringConnection.java b/src/main/java/com/rabbitmq/client/impl/recovery/AutorecoveringConnection.java
@@ -1095,8 +1095,12 @@ void recordExchange(String exchange, RecordedExchange x) {
 
     void deleteRecordedExchange(String exchange) {
         this.recordedExchanges.remove(exchange);
-        Set<RecordedBinding> xs = this.removeBindingsWithDestination(exchange);
-        for (RecordedBinding b : xs) {
+        Set<RecordedBinding> xs1 = this.removeBindingsWithDestination(exchange);
+        for (RecordedBinding b : xs1) {
+            this.maybeDeleteRecordedAutoDeleteExchange(b.getSource());
+        }
+        Set<RecordedBinding> xs2 = this.removeBindingsWithSource(exchange);
+        for (RecordedBinding b : xs2) {
             this.maybeDeleteRecordedAutoDeleteExchange(b.getSource());
         }
     }
@@ -1160,11 +1164,19 @@ boolean hasMoreConsumersOnQueue(Collection<RecordedConsumer> consumers, String q
     }
 
     Set<RecordedBinding> removeBindingsWithDestination(String s) {
+        return this.removeBindingsWithCondition(b -> b.getDestination().equals(s));
+    }
+
+    Set<RecordedBinding> removeBindingsWithSource(String s) {
+        return this.removeBindingsWithCondition(b -> b.getSource().equals(s));
+    }
+
+    private Set<RecordedBinding> removeBindingsWithCondition(Predicate<RecordedBinding> condition) {
         final Set<RecordedBinding> result = new LinkedHashSet<>();
         synchronized (this.recordedBindings) {
             for (Iterator<RecordedBinding> it = this.recordedBindings.iterator(); it.hasNext(); ) {
                 RecordedBinding b = it.next();
-                if(b.getDestination().equals(s)) {
+                if (condition.test(b)) {
                     it.remove();
                     result.add(b);
                 }
diff --git a/src/test/java/com/rabbitmq/client/test/functional/ConnectionRecovery.java b/src/test/java/com/rabbitmq/client/test/functional/ConnectionRecovery.java
@@ -865,6 +865,21 @@ public void handleDelivery(String consumerTag, Envelope envelope, BasicPropertie
         }
     }
 
+    @Test public void thatBindingFromDeletedExchangeIsDeleted() throws IOException, InterruptedException {
+        String q = generateQueueName();
+        channel.queueDeclare(q, false, false, false, null);
+        try {
+            String x = generateExchangeName();
+            channel.exchangeDeclare(x, "fanout");
+            channel.queueBind(q, x, "");
+            assertRecordedBinding(connection, 1);
+            channel.exchangeDelete(x);
+            assertRecordedBinding(connection, 0);
+        } finally {
+            channel.queueDelete(q);
+        }
+    }
+
     private void assertConsumerCount(int exp, String q) throws IOException {
         assertThat(channel.queueDeclarePassive(q).getConsumerCount()).isEqualTo(exp);
     }
@@ -1017,4 +1032,8 @@ private static void assertRecordedQueues(Connection conn, int size) {
     private static void assertRecordedExchanges(Connection conn, int size) {
         assertThat(((AutorecoveringConnection)conn).getRecordedExchanges()).hasSize(size);
     }
+
+    private static void assertRecordedBinding(Connection conn, int size) {
+        assertThat(((AutorecoveringConnection)conn).getRecordedBindings()).hasSize(size);
+    }
 }

Original file line number	Diff line number	Diff line change
`@@ -865,6 +865,21 @@ public void handleDelivery(String consumerTag, Envelope envelope, BasicPropertie`
`865`	`865`	`}`
`866`	`866`	`}`
`867`	`867`
	`868`	`+ @Test public void thatBindingFromDeletedExchangeIsDeleted() throws IOException, InterruptedException {`
	`869`	`+ String q = generateQueueName();`
	`870`	`+ channel.queueDeclare(q, false, false, false, null);`
	`871`	`+ try {`
	`872`	`+ String x = generateExchangeName();`
	`873`	`+ channel.exchangeDeclare(x, "fanout");`
	`874`	`+ channel.queueBind(q, x, "");`
	`875`	`+ assertRecordedBinding(connection, 1);`
	`876`	`+ channel.exchangeDelete(x);`
	`877`	`+ assertRecordedBinding(connection, 0);`
	`878`	`+ } finally {`
	`879`	`+ channel.queueDelete(q);`
	`880`	`+ }`
	`881`	`+ }`
	`882`	`+`
`868`	`883`	`private void assertConsumerCount(int exp, String q) throws IOException {`
`869`	`884`	`assertThat(channel.queueDeclarePassive(q).getConsumerCount()).isEqualTo(exp);`
`870`	`885`	`}`
`@@ -1017,4 +1032,8 @@ private static void assertRecordedQueues(Connection conn, int size) {`
`1017`	`1032`	`private static void assertRecordedExchanges(Connection conn, int size) {`
`1018`	`1033`	`assertThat(((AutorecoveringConnection)conn).getRecordedExchanges()).hasSize(size);`
`1019`	`1034`	`}`
	`1035`	`+`
	`1036`	`+ private static void assertRecordedBinding(Connection conn, int size) {`
	`1037`	`+ assertThat(((AutorecoveringConnection)conn).getRecordedBindings()).hasSize(size);`
	`1038`	`+ }`
`1020`	`1039`	`}`