implemented QueueDAO.containsMessage() for mysql and postgres (Netfli…

…x#1899)
s50600822 · Oct 8, 2020 · 76bf84f · 76bf84f
1 parent 75db6af
commit 76bf84f
Show file tree

Hide file tree

Showing 4 changed files with 70 additions and 0 deletions.
diff --git a/mysql-persistence/src/main/java/com/netflix/conductor/dao/mysql/MySQLQueueDAO.java b/mysql-persistence/src/main/java/com/netflix/conductor/dao/mysql/MySQLQueueDAO.java
@@ -268,4 +268,11 @@ private void createQueueIfNotExists(Connection connection, String queueName) {
 	        execute(connection, CREATE_QUEUE, q -> q.addParameter(queueName).executeUpdate());
         }
     }
+
+    @Override
+    public boolean containsMessage(String queueName, String messageId) {
+        final String EXISTS_QUEUE = "SELECT EXISTS(SELECT 1 FROM queue_message WHERE queue_name = ? AND message_id = ? )";
+        boolean exists = queryWithTransaction(EXISTS_QUEUE, q -> q.addParameter(queueName).addParameter(messageId).exists());
+        return exists;
+    }
 }
diff --git a/mysql-persistence/src/test/java/com/netflix/conductor/dao/mysql/MySQLQueueDAOTest.java b/mysql-persistence/src/test/java/com/netflix/conductor/dao/mysql/MySQLQueueDAOTest.java
@@ -104,6 +104,7 @@ public void complexQueueTest() {
 
 		for(int i = 0; i < 10; i++) {
 			String messageId = "msg" + i;
+			assertTrue(dao.containsMessage(queueName, messageId));
 			dao.remove(queueName, messageId);
 		}
 
@@ -119,6 +120,33 @@ public void complexQueueTest() {
 		assertEquals(0, size);
 	}
 
+	/**
+	 * Test fix for https://github.com/Netflix/conductor/issues/1892
+	 *
+	 * */
+	@Test
+	public void containsMessageTest() {
+		String queueName = "TestQueue";
+		long offsetTimeInSecond = 0;
+
+		for(int i = 0; i < 10; i++) {
+			String messageId = "msg" + i;
+			dao.push(queueName, messageId, offsetTimeInSecond);
+		}
+		int size = dao.getSize(queueName);
+		assertEquals(10, size);
+
+		for(int i = 0; i < 10; i++) {
+			String messageId = "msg" + i;
+			assertTrue(dao.containsMessage(queueName, messageId));
+			dao.remove(queueName, messageId);
+		}
+		for(int i = 0; i < 10; i++) {
+			String messageId = "msg" + i;
+			assertFalse(dao.containsMessage(queueName, messageId));
+		}
+	}
+
 	/**
 	 * Test fix for https://github.com/Netflix/conductor/issues/399
 	 * @since 1.8.2-rc5

diff --git a/postgres-persistence/src/main/java/com/netflix/conductor/dao/postgres/PostgresQueueDAO.java b/postgres-persistence/src/main/java/com/netflix/conductor/dao/postgres/PostgresQueueDAO.java
@@ -276,4 +276,11 @@ private void createQueueIfNotExists(Connection connection, String queueName) {
 	        execute(connection, CREATE_QUEUE, q -> q.addParameter(queueName).executeUpdate());
         }
     }
+
+    @Override
+    public boolean containsMessage(String queueName, String messageId) {
+        final String EXISTS_QUEUE = "SELECT EXISTS(SELECT 1 FROM queue_message WHERE queue_name = ? AND message_id = ? )";
+        boolean exists = queryWithTransaction(EXISTS_QUEUE, q -> q.addParameter(queueName).addParameter(messageId).exists());
+        return exists;
+    }
 }
diff --git a/...es-persistence/src/test/java/com/netflix/conductor/dao/postgres/PostgresQueueDAOTest.java b/...es-persistence/src/test/java/com/netflix/conductor/dao/postgres/PostgresQueueDAOTest.java
@@ -116,6 +116,7 @@ public void complexQueueTest() {
 
 		for(int i = 0; i < 10; i++) {
 			String messageId = "msg" + i;
+			assertTrue(dao.containsMessage(queueName, messageId));
 			dao.remove(queueName, messageId);
 		}
 
@@ -186,6 +187,33 @@ public void pollMessagesTest() {
 		}
 	}
 
+
+	/**
+	 * Test fix for https://github.com/Netflix/conductor/issues/1892
+	 *
+	 * */
+	@Test
+	public void containsMessageTest() {
+		String queueName = "TestQueue";
+		long offsetTimeInSecond = 0;
+
+		for(int i = 0; i < 10; i++) {
+			String messageId = "msg" + i;
+			dao.push(queueName, messageId, offsetTimeInSecond);
+		}
+		int size = dao.getSize(queueName);
+		assertEquals(10, size);
+
+		for(int i = 0; i < 10; i++) {
+			String messageId = "msg" + i;
+			assertTrue(dao.containsMessage(queueName, messageId));
+			dao.remove(queueName, messageId);
+		}
+		for(int i = 0; i < 10; i++) {
+			String messageId = "msg" + i;
+			assertFalse(dao.containsMessage(queueName, messageId));
+		}
+	}
 	/**
 	 * Test fix for https://github.com/Netflix/conductor/issues/448
 	 * @since 1.8.2-rc5