fix: 11348: The fix for 11231 doesn't cover ParsedBucket (#11349)

Fixes: #11348
Reviewed-by: Oleg Mazurov <oleg.mazurov@swirldslabs.com>
Signed-off-by: Artem Ananev <artem.ananev@swirldslabs.com>

platform-sdk/swirlds-jasperdb/src/main/java/com/swirlds/merkledb/files/hashmap/Bucket.java

@@ -228,7 +228,7 @@ public long findValue(final int keyHashCode, final K key, final long notFoundVal
      * @param value the entry value, this can also be special
      *     HalfDiskHashMap.INVALID_VALUE to mean delete
      */
-    public void putValue(final K key, final long value) {
+    public final void putValue(final K key, final long value) {
         putValue(key, INVALID_VALUE, value);
     }
 

platform-sdk/swirlds-jasperdb/src/main/java/com/swirlds/merkledb/files/hashmap/ParsedBucket.java

@@ -126,18 +126,20 @@ public long findValue(final int keyHashCode, final K key, final long notFoundVal
     }
 
     /**
-     * Put a key/value entry into this bucket.
-     *
-     * @param key the entry key
-     * @param value the entry value, this can also be special
-     *     HalfDiskHashMap.INVALID_VALUE to mean delete
+     * {@inheritDoc}
      */
-    public void putValue(final K key, final long value) {
+    @Override
+    public void putValue(final K key, final long oldValue, final long value) {
+        final boolean needCheckOldValue = oldValue != INVALID_VALUE;
         final int keyHashCode = key.hashCode();
         try {
             final int entryIndex = findEntryIndex(keyHashCode, key);
             if (value == INVALID_VALUE) {
-                if (entryIndex >= 0) {
+                if (entryIndex >= 0) { // if found
+                    final BucketEntry entry = entries.get(entryIndex);
+                    if (needCheckOldValue && (oldValue != entry.getValue())) {
+                        return;
+                    }
                     entries.remove(entryIndex);
                 } else {
                     // entry not found, nothing to delete
@@ -147,13 +149,18 @@ public void putValue(final K key, final long value) {
             if (entryIndex >= 0) {
                 // yay! we found it, so update value
                 final BucketEntry entry = entries.get(entryIndex);
+                if (needCheckOldValue && (oldValue != entry.getValue())) {
+                    return;
+                }
                 entry.setValue(value);
-                return;
+            } else {
+                if (needCheckOldValue) {
+                    return;
+                }
+                final BucketEntry newEntry = new BucketEntry(keyHashCode, value, key);
+                entries.add(newEntry);
+                checkLargestBucket(entries.size());
             }
-            final BucketEntry newEntry = new BucketEntry(keyHashCode, value, key);
-            entries.add(newEntry);
-
-            checkLargestBucket(entries.size());
         } catch (IOException e) {
             logger.error(EXCEPTION.getMarker(), "Failed putting key={} value={} in a bucket", key, value, e);
             throw new UncheckedIOException(e);

platform-sdk/swirlds-jasperdb/src/test/java/com/swirlds/merkledb/files/hashmap/BucketTest.java

@@ -277,6 +277,14 @@ void emptyParsedBucketToBucketIndexZero(final KeyType keyType) throws IOExceptio
         assertEquals(0, outBucket.getBucketIndex());
     }
 
+    @ParameterizedTest
+    @EnumSource(KeyType.class)
+    void parsedBucketPutIfEqual(final KeyType keyType) throws IOException {
+        final VirtualLongKey key1 = keyType.keyConstructor.apply(1L);
+        final Bucket<VirtualLongKey> bucket = new ParsedBucket<>(keyType.keySerializer, null);
+        assertDoesNotThrow(() -> bucket.putValue(key1, INVALID_VALUE, 1));
+    }
+
     private void checkKey(Bucket<VirtualLongKey> bucket, VirtualLongKey key) {
         var findResult =
                 assertDoesNotThrow(() -> bucket.findValue(key.hashCode(), key, -1), "No exception should be thrown");