Not atomic?

Author: astubbs

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/RunLengthEncoder.java

@@ -9,13 +9,12 @@
 
 import static io.confluent.csid.utils.StringUtils.msg;
 import static io.confluent.parallelconsumer.OffsetEncoding.*;
-//import static io.confluent.parallelconsumer.OffsetEncoding.RunLengthCompressed;
 
 class RunLengthEncoder extends OffsetEncoder {
 
     // todo do these need to be atomic?
-    private final AtomicInteger currentRunLengthCount;
-    private final AtomicBoolean previousRunLengthState;
+    private int currentRunLengthCount = 0;
+    private boolean previousRunLengthState = false;
 
     private final List<Integer> runLengthEncodingIntegers;
 
@@ -28,8 +27,6 @@ class RunLengthEncoder extends OffsetEncoder {
     public RunLengthEncoder(OffsetSimultaneousEncoder offsetSimultaneousEncoder, Version newVersion) {
         super(offsetSimultaneousEncoder);
         // run length setup
-        currentRunLengthCount = new AtomicInteger();
-        previousRunLengthState = new AtomicBoolean(false);
         runLengthEncodingIntegers = new ArrayList<>();
         version = newVersion;
     }
@@ -62,7 +59,7 @@ public void encodeCompletedOffset(final int rangeIndex) {
 
     @Override
     public byte[] serialise() throws EncodingNotSupportedException {
-        runLengthEncodingIntegers.add(currentRunLengthCount.get()); // add tail
+        runLengthEncodingIntegers.add(currentRunLengthCount); // add tail
 
         int entryWidth = switch (version) {
             case v1 -> Short.BYTES;
@@ -81,7 +78,8 @@ public byte[] serialise() throws EncodingNotSupportedException {
                 case v2 -> {
                     runLengthEncodedByteBuffer.putInt(runlength);
                 }
-            };
+            }
+            ;
         }
 
         byte[] array = runLengthEncodedByteBuffer.array();
@@ -101,13 +99,13 @@ protected byte[] getEncodedBytes() {
 
     private void encodeRunLength(final boolean currentIsComplete) {
         // run length
-        boolean currentOffsetMatchesOurRunLengthState = previousRunLengthState.get() == currentIsComplete;
+        boolean currentOffsetMatchesOurRunLengthState = previousRunLengthState == currentIsComplete;
         if (currentOffsetMatchesOurRunLengthState) {
-            currentRunLengthCount.getAndIncrement();
+            currentRunLengthCount++;
         } else {
-            previousRunLengthState.set(currentIsComplete);
-            runLengthEncodingIntegers.add(currentRunLengthCount.get());
-            currentRunLengthCount.set(1); // reset to 1
+            previousRunLengthState = currentIsComplete;
+            runLengthEncodingIntegers.add(currentRunLengthCount);
+            currentRunLengthCount = 1; // reset to 1
         }
     }
 }