@Test public void testIdSchemaCaching() throws IOException, SchemaRegistryException { KafkaSchemaRegistry<Integer, String> baseRegistry = mock(KafkaSchemaRegistry.class); String name = "test"; String schema1 = new String("schema"); Integer id1 = 1; CachingKafkaSchemaRegistry<Integer, String> cachingReg = new CachingKafkaSchemaRegistry<>(baseRegistry, 2); when(baseRegistry.getById(id1)).thenReturn(schema1); String schemaReturned = cachingReg.getById(id1); Assert.assertEquals(schemaReturned, schema1, "Schema returned by id should be the same"); verify(baseRegistry, times(1)).getById(anyInt()); when(baseRegistry.getById(id1)).thenReturn(new String("schema2")); Assert.assertEquals(cachingReg.getById(id1), schemaReturned); verify(baseRegistry, times(1)).getById(anyInt()); }
@Override synchronized public S getById(K id) throws IOException, SchemaRegistryException { if (_idBasedCache.containsKey(id)) { return _idBasedCache.get(id); } else { S schema = _kafkaSchemaRegistry.getById(id); _idBasedCache.put(id, schema); return schema; } }
CachingKafkaSchemaRegistry implements KafkaSchemaRegistry<K,S> { @Override synchronized public S getById(K id) throws IOException, SchemaRegistryException { if (_idBasedCache.containsKey(id)) { return _idBasedCache.get(id); } else { S schema = _kafkaSchemaRegistry.getById(id); _idBasedCache.put(id, schema); return schema; } } }
CachingKafkaSchemaRegistry implements KafkaSchemaRegistry<K,S> { @Override synchronized public S getById(K id) throws IOException, SchemaRegistryException { if (_idBasedCache.containsKey(id)) { return _idBasedCache.get(id); } else { S schema = _kafkaSchemaRegistry.getById(id); _idBasedCache.put(id, schema); return schema; } } CachingKafkaSchemaRegistry(KafkaSchemaRegistry kafkaSchemaRegistry);  CachingKafkaSchemaRegistry(KafkaSchemaRegistry kafkaSchemaRegistry, int maxSchemaReferences); }
CachingKafkaSchemaRegistry implements KafkaSchemaRegistry<K,S> { @Override synchronized public S getById(K id) throws IOException, SchemaRegistryException { if (_idBasedCache.containsKey(id)) { return _idBasedCache.get(id); } else { S schema = _kafkaSchemaRegistry.getById(id); _idBasedCache.put(id, schema); return schema; } } CachingKafkaSchemaRegistry(KafkaSchemaRegistry kafkaSchemaRegistry);  CachingKafkaSchemaRegistry(KafkaSchemaRegistry kafkaSchemaRegistry, int maxSchemaReferences); @Override synchronized K register(String name, S schema); @Override synchronized S getById(K id); @Override S getLatestSchema(String name); @Override boolean hasInternalCache(); }
CachingKafkaSchemaRegistry implements KafkaSchemaRegistry<K,S> { @Override synchronized public S getById(K id) throws IOException, SchemaRegistryException { if (_idBasedCache.containsKey(id)) { return _idBasedCache.get(id); } else { S schema = _kafkaSchemaRegistry.getById(id); _idBasedCache.put(id, schema); return schema; } } CachingKafkaSchemaRegistry(KafkaSchemaRegistry kafkaSchemaRegistry);  CachingKafkaSchemaRegistry(KafkaSchemaRegistry kafkaSchemaRegistry, int maxSchemaReferences); @Override synchronized K register(String name, S schema); @Override synchronized S getById(K id); @Override S getLatestSchema(String name); @Override boolean hasInternalCache();  }