@Test public void testMaxReferences() throws IOException, SchemaRegistryException { KafkaSchemaRegistry<Integer, String> baseRegistry = mock(KafkaSchemaRegistry.class); String name = "test"; String schema1 = new String("schema"); String schema2 = new String("schema"); String schema3 = new String("schema"); Integer id1 = 1; Integer id2 = 2; Integer id3 = 3; CachingKafkaSchemaRegistry<Integer, String> cachingReg = new CachingKafkaSchemaRegistry<>(baseRegistry, 2); when(baseRegistry.register(name, schema1)).thenReturn(id1); Assert.assertEquals(cachingReg.register(name, schema1), id1); when(baseRegistry.register(name, schema2)).thenReturn(id2); Assert.assertEquals(cachingReg.register(name, schema2), id2); when(baseRegistry.register(name, schema3)).thenReturn(id3); try { cachingReg.register(name, schema3); Assert.fail("Should have thrown an exception"); } catch (Exception e) { log.info(e.getMessage()); } }
@Override synchronized public K register(String name, S schema) throws IOException, SchemaRegistryException { Map<S, K> schemaIdMap; if (_namedSchemaCache.containsKey(name)) { schemaIdMap = _namedSchemaCache.get(name); } else { schemaIdMap = new IdentityHashMap<>(); _namedSchemaCache.put(name, schemaIdMap); } if (schemaIdMap.containsKey(schema)) { return schemaIdMap.get(schema); } else { Preconditions.checkState(schemaIdMap.size() < _maxSchemaReferences, "Too many schema objects for " + name +". Cache is overfull."); } K id = _kafkaSchemaRegistry.register(name, schema); schemaIdMap.put(schema, id); _idBasedCache.put(id, schema); return id; }
CachingKafkaSchemaRegistry implements KafkaSchemaRegistry<K,S> { @Override synchronized public K register(String name, S schema) throws IOException, SchemaRegistryException { Map<S, K> schemaIdMap; if (_namedSchemaCache.containsKey(name)) { schemaIdMap = _namedSchemaCache.get(name); } else { schemaIdMap = new IdentityHashMap<>(); _namedSchemaCache.put(name, schemaIdMap); } if (schemaIdMap.containsKey(schema)) { return schemaIdMap.get(schema); } else { Preconditions.checkState(schemaIdMap.size() < _maxSchemaReferences, "Too many schema objects for " + name +". Cache is overfull."); } K id = _kafkaSchemaRegistry.register(name, schema); schemaIdMap.put(schema, id); _idBasedCache.put(id, schema); return id; } }
CachingKafkaSchemaRegistry implements KafkaSchemaRegistry<K,S> { @Override synchronized public K register(String name, S schema) throws IOException, SchemaRegistryException { Map<S, K> schemaIdMap; if (_namedSchemaCache.containsKey(name)) { schemaIdMap = _namedSchemaCache.get(name); } else { schemaIdMap = new IdentityHashMap<>(); _namedSchemaCache.put(name, schemaIdMap); } if (schemaIdMap.containsKey(schema)) { return schemaIdMap.get(schema); } else { Preconditions.checkState(schemaIdMap.size() < _maxSchemaReferences, "Too many schema objects for " + name +". Cache is overfull."); } K id = _kafkaSchemaRegistry.register(name, schema); schemaIdMap.put(schema, id); _idBasedCache.put(id, schema); return id; } CachingKafkaSchemaRegistry(KafkaSchemaRegistry kafkaSchemaRegistry);  CachingKafkaSchemaRegistry(KafkaSchemaRegistry kafkaSchemaRegistry, int maxSchemaReferences); }
CachingKafkaSchemaRegistry implements KafkaSchemaRegistry<K,S> { @Override synchronized public K register(String name, S schema) throws IOException, SchemaRegistryException { Map<S, K> schemaIdMap; if (_namedSchemaCache.containsKey(name)) { schemaIdMap = _namedSchemaCache.get(name); } else { schemaIdMap = new IdentityHashMap<>(); _namedSchemaCache.put(name, schemaIdMap); } if (schemaIdMap.containsKey(schema)) { return schemaIdMap.get(schema); } else { Preconditions.checkState(schemaIdMap.size() < _maxSchemaReferences, "Too many schema objects for " + name +". Cache is overfull."); } K id = _kafkaSchemaRegistry.register(name, schema); schemaIdMap.put(schema, id); _idBasedCache.put(id, schema); return id; } 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 K register(String name, S schema) throws IOException, SchemaRegistryException { Map<S, K> schemaIdMap; if (_namedSchemaCache.containsKey(name)) { schemaIdMap = _namedSchemaCache.get(name); } else { schemaIdMap = new IdentityHashMap<>(); _namedSchemaCache.put(name, schemaIdMap); } if (schemaIdMap.containsKey(schema)) { return schemaIdMap.get(schema); } else { Preconditions.checkState(schemaIdMap.size() < _maxSchemaReferences, "Too many schema objects for " + name +". Cache is overfull."); } K id = _kafkaSchemaRegistry.register(name, schema); schemaIdMap.put(schema, id); _idBasedCache.put(id, schema); return id; } 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();  }