From 8a515bd91935e120195a38c351c2c51328101e25 Mon Sep 17 00:00:00 2001
From: Shichao Nie <niesc@automq.com>
Date: Tue, 2 Jul 2024 20:37:20 +0800
Subject: [PATCH] fix(metadata): prevent duplicate release for constant
 instance

Signed-off-by: Shichao Nie <niesc@automq.com>
---
 .../org/apache/kafka/image/RegistryRef.java   | 33 +++++++++++++++++--
 .../apache/kafka/image/S3ObjectsImage.java    |  3 ++
 .../kafka/image/S3StreamsMetadataImage.java   |  3 ++
 3 files changed, 37 insertions(+), 2 deletions(-)

diff --git a/metadata/src/main/java/org/apache/kafka/image/RegistryRef.java b/metadata/src/main/java/org/apache/kafka/image/RegistryRef.java
index f1d837ce22..41739bc0ca 100644
--- a/metadata/src/main/java/org/apache/kafka/image/RegistryRef.java
+++ b/metadata/src/main/java/org/apache/kafka/image/RegistryRef.java
@@ -54,10 +54,39 @@ public RegistryRef next() {
     }
 
     @Override
-    protected void deallocate() {
+    public ReferenceCounted retain(int increment) {
         if (this == NOOP) {
-            return;
+            throw new UnsupportedOperationException("retain is not supported for NOOP");
         }
+        return super.retain(increment);
+    }
+
+    @Override
+    public ReferenceCounted retain() {
+        if (this == NOOP) {
+            throw new UnsupportedOperationException("retain is not supported for NOOP");
+        }
+        return super.retain();
+    }
+
+    @Override
+    public boolean release() {
+        if (this == NOOP) {
+            throw new UnsupportedOperationException("release is not supported for NOOP");
+        }
+        return super.release();
+    }
+
+    @Override
+    public boolean release(int decrement) {
+        if (this == NOOP) {
+            throw new UnsupportedOperationException("release is not supported for NOOP");
+        }
+        return super.release(decrement);
+    }
+
+    @Override
+    protected void deallocate() {
         inLock(() -> {
             if (liveEpochs.isEmpty()) {
                 throw new IllegalStateException("liveEpochs is empty");
diff --git a/metadata/src/main/java/org/apache/kafka/image/S3ObjectsImage.java b/metadata/src/main/java/org/apache/kafka/image/S3ObjectsImage.java
index f50d13082d..46ee662003 100644
--- a/metadata/src/main/java/org/apache/kafka/image/S3ObjectsImage.java
+++ b/metadata/src/main/java/org/apache/kafka/image/S3ObjectsImage.java
@@ -139,6 +139,9 @@ RegistryRef registryRef() {
 
     @Override
     protected void deallocate() {
+        if (registryRef == RegistryRef.NOOP) {
+            return;
+        }
         registryRef.release();
     }
 
diff --git a/metadata/src/main/java/org/apache/kafka/image/S3StreamsMetadataImage.java b/metadata/src/main/java/org/apache/kafka/image/S3StreamsMetadataImage.java
index d8cd273583..2990e07aa6 100644
--- a/metadata/src/main/java/org/apache/kafka/image/S3StreamsMetadataImage.java
+++ b/metadata/src/main/java/org/apache/kafka/image/S3StreamsMetadataImage.java
@@ -408,6 +408,9 @@ public String toString() {
 
     @Override
     protected void deallocate() {
+        if (registryRef == RegistryRef.NOOP) {
+            return;
+        }
         registryRef.release();
     }