Fix broken throttler flush for v0.11.1.

Author: vy

CHANGELOG.md

@@ -14,6 +14,10 @@
  limitations under the License.
 -->
 
+### (2022-10-04) v0.11.1
+
+- Earlier throttler fix has introduced a bug. The buffer cursor is not reset correctly and caused duplicates in logged events. (Thanks to [Khaled Bakhtiari](https://github.com/ec84b4) for the report.)
+
 ### (2022-09-02) v0.11.0
 
 - Flush remaining events in the buffer after interrupting the throttler.

README.md

@@ -246,9 +246,10 @@ F.A.Q.
 Contributors
 ============
 
+- [Boris Faniuk](https://github.com//bfanyuk)
+- [Khaled Bakhtiari](https://github.com/ec84b4)
 - [t9t](https://github.com/t9t)
 - [Yaroslav Skopets](https://github.com/yskopets)
-- [Boris Faniuk](https://github.com/bfanyuk)
 
 # License
 

appender/src/main/java/com/vlkan/log4j2/redis/appender/RedisThrottler.java

@@ -219,6 +219,7 @@ private void flush(long waitPeriodNanos) throws InterruptedException {
             LOGGER.debug("{} background task is pushing last {} events that didn't fit into the batch", logPrefix, batchIndex);
             byte[][] subBatch = Arrays.copyOfRange(batch, 0, batchIndex);
             push(subBatch);
+            batchIndex = 0;
         }
 
     }

appender/src/test/java/com/vlkan/log4j2/redis/appender/RedisAppenderTest.java

@@ -75,16 +75,56 @@ void appended_messages_should_be_persisted() {
             logger.log(expectedLogMessage.level, expectedLogMessage.message);
         }
 
+        // Verify the logging.
+        verifyLogging(expectedLogMessages, expectedMessageCount, expectedMessageCount);
+
+    }
+
+    @Test
+    void throttler_should_not_flush_same_content_twice() {
+
+        // Create the logger.
+        LOGGER.debug("creating the logger");
+        Logger logger = loggerContextExtension
+                .getLoggerContext()
+                .getLogger(RedisAppenderTest.class.getCanonicalName());
+
+        // Verify that batch size is greater than 1 so that we can observe a partially filled batch push.
+        Assertions.assertThat(RedisAppenderTestConfig.THROTTLER_BATCH_SIZE).isGreaterThan(1);
+
+        // Log the 1st message.
+        RedisTestMessage[] expectedLogMessages1 = RedisTestMessage.createRandomArray(1);
+        logger.log(expectedLogMessages1[0].level, expectedLogMessages1[0].message);
+
+        // Verify the 1st message persistence.
+        verifyLogging(expectedLogMessages1, 1, 1);
+
+        // Log the 1st message.
+        RedisTestMessage[] expectedLogMessages2 = RedisTestMessage.createRandomArray(1);
+        logger.log(expectedLogMessages2[0].level, expectedLogMessages2[0].message);
+
+        // Verify the 1st message persistence.
+        verifyLogging(expectedLogMessages2, 2, 2);
+
+    }
+
+    private void verifyLogging(
+            RedisTestMessage[] expectedLogMessages,
+            int expectedTotalEventCount,
+            int expectedRedisPushSuccessCount) {
+
         // Verify the amount of persisted messages.
         Jedis jedis = redisClientExtension.getClient();
         LOGGER.debug("waiting for the logged messages to be persisted");
+        int expectedMessageCount = expectedLogMessages.length;
         Awaitility
                 .await()
                 .pollDelay(Duration.ofMillis(100))
                 .atMost(Duration.ofSeconds(10))
                 .until(() -> {
                     long persistedMessageCount = jedis.llen(RedisAppenderTestConfig.REDIS_KEY);
-                    return persistedMessageCount == expectedLogMessages.length;
+                    Assertions.assertThat(persistedMessageCount).isLessThanOrEqualTo(expectedMessageCount);
+                    return persistedMessageCount == expectedMessageCount;
                 });
 
         // Verify the content of persisted messages.
@@ -115,12 +155,12 @@ void appended_messages_should_be_persisted() {
                 .getAppender(RedisAppenderTestConfig.LOG4J2_APPENDER_NAME);
         Assertions.assertThat(appender).isInstanceOf(RedisAppender.class);
         RedisThrottlerJmxBean jmxBean = ((RedisAppender) appender).getJmxBean();
-        Assertions.assertThat(jmxBean.getTotalEventCount()).isEqualTo(expectedMessageCount);
+        Assertions.assertThat(jmxBean.getTotalEventCount()).isEqualTo(expectedTotalEventCount);
         Assertions.assertThat(jmxBean.getIgnoredEventCount()).isEqualTo(0);
         Assertions.assertThat(jmxBean.getEventRateLimitFailureCount()).isEqualTo(0);
         Assertions.assertThat(jmxBean.getByteRateLimitFailureCount()).isEqualTo(0);
         Assertions.assertThat(jmxBean.getUnavailableBufferSpaceFailureCount()).isEqualTo(0);
-        Assertions.assertThat(jmxBean.getRedisPushSuccessCount()).isEqualTo(expectedMessageCount);
+        Assertions.assertThat(jmxBean.getRedisPushSuccessCount()).isEqualTo(expectedRedisPushSuccessCount);
         Assertions.assertThat(jmxBean.getRedisPushFailureCount()).isEqualTo(0);
 
     }

appender/src/test/java/com/vlkan/log4j2/redis/appender/RedisAppenderTestConfig.java

@@ -34,4 +34,6 @@ enum RedisAppenderTestConfig {;
 
     static final String LOG4J2_APPENDER_NAME = "REDIS";
 
+    static final int THROTTLER_BATCH_SIZE = 10;
+
 }