Fix queue invariant checks

iotdb-core/calc-commons/pom.xml

@@ -99,6 +99,11 @@
             <groupId>at.yawk.lz4</groupId>
             <artifactId>lz4-java</artifactId>
         </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
     <build>
         <plugins>

iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/schedule/queue/IndexedBlockingQueue.java

@@ -42,6 +42,8 @@ public abstract class IndexedBlockingQueue<E extends IDIndexedAccessible> {
 
   public static final String TOO_MANY_CONCURRENT_QUERIES_ERROR_MSG =
       "The system can't allow more queries.";
+  public static final String SAME_ID_ELEMENT_ALREADY_EXISTS_ERROR_MSG =
+      "The queue has already contained the same ID element.";
 
   protected final int capacity;
   protected final E queryHolder;
@@ -57,6 +59,7 @@ public abstract class IndexedBlockingQueue<E extends IDIndexedAccessible> {
    * @throws IllegalArgumentException if maxCapacity <= 0.
    */
   protected IndexedBlockingQueue(int maxCapacity, E queryHolder) {
+    Preconditions.checkArgument(maxCapacity > 0, "maxCapacity must be greater than 0.");
     this.capacity = maxCapacity;
     this.queryHolder = queryHolder;
   }
@@ -93,6 +96,7 @@ public synchronized void push(E element) {
       throw new NullPointerException(CalcMessages.PUSHED_ELEMENT_IS_NULL);
     }
     Preconditions.checkState(size < capacity, TOO_MANY_CONCURRENT_QUERIES_ERROR_MSG);
+    Preconditions.checkState(!contains(element), SAME_ID_ELEMENT_ALREADY_EXISTS_ERROR_MSG);
     pushToQueue(element);
     size++;
     this.notifyAll();

iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/schedule/queue/IndexedBlockingReserveQueue.java

@@ -57,6 +57,7 @@ public synchronized void push(E element) {
       throw new NullPointerException(CalcMessages.PUSHED_ELEMENT_IS_NULL);
     }
     Preconditions.checkState(size + reservedSize < capacity, TOO_MANY_CONCURRENT_QUERIES_ERROR_MSG);
+    Preconditions.checkState(!contains(element), SAME_ID_ELEMENT_ALREADY_EXISTS_ERROR_MSG);
     pushToQueue(element);
     size++;
     this.notifyAll();
@@ -67,6 +68,8 @@ public synchronized void repush(E element) {
     if (element == null) {
       throw new NullPointerException(CalcMessages.PUSHED_ELEMENT_IS_NULL);
     }
+    Preconditions.checkState(reservedSize > 0, "No reserved space is available.");
+    Preconditions.checkState(!contains(element), SAME_ID_ELEMENT_ALREADY_EXISTS_ERROR_MSG);
     pushToQueue(element);
     reservedSize--;
     size++;
@@ -77,6 +80,13 @@ public synchronized void repush(E element) {
    * For task that is not in readyQueue when it's cleared, it won't be added into the queue again.
    */
   public synchronized void decreaseReservedSize() {
+    Preconditions.checkState(reservedSize > 0, "No reserved space is available.");
     this.reservedSize--;
   }
+
+  @Override
+  public synchronized void clear() {
+    super.clear();
+    reservedSize = 0;
+  }
 }

iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/schedule/queue/IndexedBlockingQueueTest.java

@@ -0,0 +1,241 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.calc.execution.schedule.queue;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.ArrayDeque;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Queue;
+
+public class IndexedBlockingQueueTest {
+
+  @Test
+  public void testRejectNonPositiveCapacity() {
+    Assert.assertThrows(IllegalArgumentException.class, () -> new SimpleQueue(0));
+    Assert.assertThrows(IllegalArgumentException.class, () -> new ReserveQueue(-1));
+  }
+
+  @Test
+  public void testRejectDuplicateIdPush() throws InterruptedException {
+    SimpleQueue queue = new SimpleQueue(2);
+    Element first = new Element(1);
+    Element duplicate = new Element(1);
+
+    queue.push(first);
+    Assert.assertThrows(IllegalStateException.class, () -> queue.push(duplicate));
+    Assert.assertEquals(1, queue.size());
+    Assert.assertSame(first, queue.poll());
+    Assert.assertTrue(queue.isEmpty());
+  }
+
+  @Test
+  public void testRejectRepushWithoutReservedSpace() {
+    ReserveQueue queue = new ReserveQueue(1);
+
+    Assert.assertThrows(IllegalStateException.class, () -> queue.repush(new Element(1)));
+  }
+
+  @Test
+  public void testRejectDecreaseWithoutReservedSpace() {
+    ReserveQueue queue = new ReserveQueue(1);
+
+    Assert.assertThrows(IllegalStateException.class, queue::decreaseReservedSize);
+  }
+
+  @Test
+  public void testReservedSpaceCanOnlyBeReleasedOnce() throws InterruptedException {
+    ReserveQueue queue = new ReserveQueue(1);
+    Element element = new Element(1);
+    queue.push(element);
+
+    Assert.assertSame(element, queue.poll());
+    queue.decreaseReservedSize();
+
+    Assert.assertThrows(IllegalStateException.class, queue::decreaseReservedSize);
+  }
+
+  @Test
+  public void testClearReleasesReservedSpace() throws InterruptedException {
+    ReserveQueue queue = new ReserveQueue(1);
+    Element first = new Element(1);
+    Element second = new Element(2);
+
+    queue.push(first);
+    Assert.assertSame(first, queue.poll());
+    queue.clear();
+
+    queue.push(second);
+    Assert.assertEquals(1, queue.size());
+    Assert.assertSame(second, queue.poll());
+  }
+
+  private static class SimpleQueue extends IndexedBlockingQueue<Element> {
+    private final Queue<Element> elements = new ArrayDeque<>();
+    private final Map<ID, Element> keyedElements = new HashMap<>();
+
+    private SimpleQueue(int capacity) {
+      super(capacity, new Element(0));
+    }
+
+    @Override
+    protected Element remove(Element element) {
+      Element removed = keyedElements.remove(element.getDriverTaskId());
+      if (removed != null) {
+        elements.remove(removed);
+      }
+      return removed;
+    }
+
+    @Override
+    protected Element get(Element element) {
+      return keyedElements.get(element.getDriverTaskId());
+    }
+
+    @Override
+    public boolean isEmpty() {
+      return elements.isEmpty();
+    }
+
+    @Override
+    protected Element pollFirst() {
+      Element first = elements.remove();
+      keyedElements.remove(first.getDriverTaskId());
+      return first;
+    }
+
+    @Override
+    protected void pushToQueue(Element element) {
+      elements.add(element);
+      keyedElements.put(element.getDriverTaskId(), element);
+    }
+
+    @Override
+    protected boolean contains(Element element) {
+      return keyedElements.containsKey(element.getDriverTaskId());
+    }
+
+    @Override
+    protected void clearAllElements() {
+      elements.clear();
+      keyedElements.clear();
+    }
+  }
+
+  private static class ReserveQueue extends IndexedBlockingReserveQueue<Element> {
+    private final Queue<Element> elements = new ArrayDeque<>();
+    private final Map<ID, Element> keyedElements = new HashMap<>();
+
+    private ReserveQueue(int capacity) {
+      super(capacity, new Element(0));
+    }
+
+    @Override
+    protected Element remove(Element element) {
+      Element removed = keyedElements.remove(element.getDriverTaskId());
+      if (removed != null) {
+        elements.remove(removed);
+      }
+      return removed;
+    }
+
+    @Override
+    protected Element get(Element element) {
+      return keyedElements.get(element.getDriverTaskId());
+    }
+
+    @Override
+    public boolean isEmpty() {
+      return elements.isEmpty();
+    }
+
+    @Override
+    protected Element pollFirst() {
+      Element first = elements.remove();
+      keyedElements.remove(first.getDriverTaskId());
+      return first;
+    }
+
+    @Override
+    protected void pushToQueue(Element element) {
+      elements.add(element);
+      keyedElements.put(element.getDriverTaskId(), element);
+    }
+
+    @Override
+    protected boolean contains(Element element) {
+      return keyedElements.containsKey(element.getDriverTaskId());
+    }
+
+    @Override
+    protected void clearAllElements() {
+      elements.clear();
+      keyedElements.clear();
+    }
+  }
+
+  private static class Element implements IDIndexedAccessible {
+    private ElementId id;
+
+    private Element(int id) {
+      this.id = new ElementId(id);
+    }
+
+    @Override
+    public ID getDriverTaskId() {
+      return id;
+    }
+
+    @Override
+    public void setId(ID id) {
+      this.id = (ElementId) id;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+      return obj instanceof Element && ((Element) obj).id.equals(id);
+    }
+
+    @Override
+    public int hashCode() {
+      return id.hashCode();
+    }
+  }
+
+  private static class ElementId implements ID {
+    private final int id;
+
+    private ElementId(int id) {
+      this.id = id;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+      return obj instanceof ElementId && ((ElementId) obj).id == id;
+    }
+
+    @Override
+    public int hashCode() {
+      return Integer.hashCode(id);
+    }
+  }
+}

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigRegionListener.java

@@ -50,6 +50,9 @@ public synchronized void increaseReference(final PipeParameters parameters)
   public synchronized void decreaseReference(final PipeParameters parameters)
       throws IllegalPathException {
     if (!ConfigRegionListeningFilter.parseListeningPlanTypeSet(parameters).isEmpty()) {
+      if (listeningQueueReferenceCount == 0) {
+        return;
+      }
       listeningQueueReferenceCount--;
       if (listeningQueueReferenceCount == 0) {
         listeningQueue.close();