Skip to content

Fix/codec registry thread safety #699

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
alex268 wants to merge 2 commits into
base: master
Choose a base branch
from
+22 −16
Open

Fix/codec registry thread safety #699

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
87390cb
Making CodecRegistry safe for concurrent use
Aug 4, 2026
a4a67f1
Updated CodecRegistryTest
alex268 Aug 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 7 additions & 5 deletions topic/src/main/java/tech/ydb/topic/description/CodecRegistry.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
package tech.ydb.topic.description;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand All @@ -17,10 +17,11 @@ public class CodecRegistry {

private static final Logger logger = LoggerFactory.getLogger(CodecRegistry.class);

final Map<Integer, Codec> customCodecMap;
// registerCodec may be called at any moment, while getCodec is used by the compression and the
// decompression threads of every reader and writer created from the same TopicClient
private final Map<Integer, Codec> customCodecMap = new ConcurrentHashMap<>();

@robot-vibe-db robot-vibe-db Bot Aug 12, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Severity: Major
Confidence: High

The concurrent test registerCustomCodecIsSafeForConcurrentUse — added in the first commit (87390cb) specifically to validate this HashMap→ConcurrentHashMap change — was removed in the second commit (a4a67f1). The first commit's message states: "The new test loses codecs on every run with the old implementation."

This test is the only automated validation that the thread-safety fix works. Without it, there is no regression test to catch future changes that might reintroduce the race (e.g., someone refactoring back to a plain HashMap, or introducing a compound read-then-write operation that ConcurrentHashMap alone doesn't make atomic).

Suggested fix: restore the registerCustomCodecIsSafeForConcurrentUse test. It uses 4 writer threads + 1 reader thread with a CountDownLatch to stress-test concurrent registrations, and asserts that no codecs are lost — this is exactly the kind of test that should stay as a regression guard.


public CodecRegistry() {
customCodecMap = new HashMap<>();
for (Codec codec: StandardCodecs.getAvailableCodecs()) {
customCodecMap.put(codec.getId(), codec);
}
Expand All @@ -32,11 +33,12 @@ public CodecRegistry() {
* @return previous implementation with associated codec
*/
public Codec registerCodec(Codec codec) {
assert codec != null;
if (codec == null) {
throw new IllegalArgumentException("Codec must be not null");
}
int codecId = codec.getId();

Codec result = customCodecMap.put(codecId, codec);

if (result != null) {
logger.info(
"Replace codec which have already associated with this id. CodecId: {} Codec: {}",
Expand Down
26 changes: 15 additions & 11 deletions topic/src/test/java/tech/ydb/topic/impl/CodecRegistryTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,16 +1,16 @@
package tech.ydb.topic.impl;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

import tech.ydb.topic.description.Codec;
import tech.ydb.topic.description.CodecRegistry;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

/**
* Unit tests for check simple logic for register custom codec
*
Expand All @@ -19,7 +19,7 @@
public class CodecRegistryTest {
CodecRegistry registry;

private static final int codecId = 10113;
private static final int CUSTOM_ID = 10113;

@Before
public void beforeTest() {
Expand All @@ -34,15 +34,14 @@ public void registerCustomCodecShouldDoubleRegisterCodecAndReturnLastCodec() {
registry.registerCodec(codec1);
Assert.assertEquals(codec1, registry.registerCodec(codec2));

Assert.assertEquals(codec2, registry.getCodec(codecId));
Assert.assertNotEquals(codec1, registry.getCodec(codecId));
Assert.assertEquals(codec2, registry.getCodec(CUSTOM_ID));
Assert.assertNotEquals(codec1, registry.getCodec(CUSTOM_ID));
}

@Test
public void registerCustomCodecShouldNotAcceptNull() {
Assert.assertThrows(
AssertionError.class,
() -> registry.registerCodec(null));
Exception ex = Assert.assertThrows(IllegalArgumentException.class, () -> registry.registerCodec(null));
Assert.assertEquals("Codec must be not null", ex.getMessage());
}

@Test
Expand All @@ -65,7 +64,12 @@ static class CodecTopic implements Codec {
int codec;

public CodecTopic() {
this.codec = codecId;
this.codec = CUSTOM_ID;
}

@Override
public String toString() {
return "CustomCodec";
}

public void setCodecId(int codecId) {
Expand Down
Loading