Kafka Streams 1.0.0状态存储读取行为及性能优化咨询

First off, let's clarify the core question you're asking: When you call org.apache.kafka.streams.state.ReadOnlyKeyValueStore.get(String key) on a global table, you are NOT reading from Kafka every time. Global KTables sync their entire dataset to local state stores (default is RocksDB, with in-memory options available) upfront, and all get() operations hit this local storage directly.

So why is your get() call taking up such a high percentage of execution time? Let's look at the most likely culprits:

• Suboptimal state store configuration: If you're using the default RocksDB store, it might not be tuned for your workload—like insufficient memory allocation leading to frequent disk I/O, or missing optimizations like prefix compression.
• Inefficient serialization/deserialization: If your List<String> value is being serialized with a slow serializer (like naive JSON or unoptimized custom code), every get() requires expensive deserialization that adds up.
• Hot key traffic: If certain customerCode values are queried extremely frequently, repeated disk access (for RocksDB) can amplify latency.
• Global table sync delays: While less common, if your global table is still syncing the full dataset from Kafka, you might see temporary delays—but this usually resolves once the initial sync completes.

Fixes to Optimize get() Performance #

1. Tune Your State Store Configuration #

For RocksDB (the default persistent store), adjust settings to reduce disk I/O and improve cache utilization:

StreamsBuilder builder = new StreamsBuilder();

// Custom RocksDB config to boost performance
Map<String, String> storeConfigs = new HashMap<>();
// Increase block cache size to reduce disk reads (64MB example)
storeConfigs.put(StreamsConfig.ROCKSDB_CONFIG_SETTING, "block_cache_size=67108864");
// Enable prefix compression to reduce storage footprint and improve read speed
storeConfigs.put(StreamsConfig.ROCKSDB_CONFIG_SETTING, "prefix_extractor=org.rocksdb.PrefixLengthPrefixExtractor(10)");

// Attach config when creating your global table
GlobalKTable<String, List<String>> policyGlobalTable = builder.globalTable(
    "policy-topic",
    Materialized.<String, List<String>, KeyValueStore<Bytes, byte[]>>as(policyGlobalTableName)
        .withKeySerde(Serdes.String())
        .withValueSerde(new OptimizedListStringSerde())
        .withStoreConfig(storeConfigs)
);

If your dataset is small enough, switch to an in-memory state store to eliminate disk I/O entirely:

Materialized.<String, List<String>, KeyValueStore<Bytes, byte[]>>as(policyGlobalTableName)
    .withStoreType(KeyValueStore.MEMORY_STORE_NAME)

2. Optimize Serialization/Deserialization #

Replace slow serializers with a custom, efficient implementation for your List<String> value. For example, use a compact binary format or optimized string encoding:

public class OptimizedListStringSerde extends Serde<List<String>> {
    @Override
    public Serializer<List<String>> serializer() {
        return (topic, list) -> {
            if (list == null) return null;
            // Use a delimiter that won't appear in your data for fast splitting
            return String.join("|||", list).getBytes(StandardCharsets.UTF_8);
        };
    }

    @Override
    public Deserializer<List<String>> deserializer() {
        return (topic, bytes) -> {
            if (bytes == null) return Collections.emptyList();
            return Arrays.asList(new String(bytes, StandardCharsets.UTF_8).split("\\|\\|\\|"));
        };
    }
}

3. Add a Local Cache for Hot Keys #

If specific customerCode values are queried constantly, add an application-level cache (like Guava Cache) to reduce hits to the state store:

// Initialize a cache with size limits and expiration to keep data fresh
LoadingCache<String, List<String>> policyCache = CacheBuilder.newBuilder()
    .maximumSize(1000) // Cache top 1000 hot keys
    .expireAfterWrite(5, TimeUnit.MINUTES) // Refresh cache every 5 minutes
    .build(new CacheLoader<String, List<String>>() {
        @Override
        public List<String> load(String customerCode) {
            // Fall back to state store when cache misses
            return policyStore.get(customerCode);
        }
    });

// Use the cache instead of direct state store calls
List<String> policyIds = policyCache.get(event.getCustomerCode());

4. Verify Global Table Sync Status #

Check if your global table has finished initial sync using Kafka Streams metrics (like kafka_streams_global_table_offset_lag). If lag is high, adjust your Streams application's thread count or check the source topic's partition distribution to speed up syncing.

Final Recap #

• get() operations on a global table always hit local storage—never Kafka directly.
• High latency from get() is almost always due to state store tuning, serialization, or hot keys.
• Implementing the fixes above should drastically reduce the time spent on these calls.

内容的提问来源于stack exchange，提问作者Rednam Nagendra