@Test public void testGetAlwaysTrueHostnameVerifierWhenSslEndpointIdentificationAlgorithmIsNotSet() throws Exception { Properties props = new Properties(); props.put(SchemaRegistryConfig.KAFKASTORE_BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); props.put(SchemaRegistryConfig.KAFKASTORE_TOPIC_CONFIG, ClusterTestHarness.KAFKASTORE_TOPIC); SchemaRegistryConfig config = new SchemaRegistryConfig(props); KafkaSchemaRegistry schemaRegistry = new KafkaSchemaRegistry( config, new SchemaRegistrySerializer() ); assertTrue(schemaRegistry.getHostnameVerifier().verify("", null)); }
@Override public V put(K key, V value) throws StoreTimeoutException, StoreException { assertInitialized(); if (key == null) { throw new StoreException("Key should not be null"); } V oldValue = get(key); ProducerRecord<byte[], byte[]> producerRecord = null; try { producerRecord = new ProducerRecord<byte[], byte[]>(topic, 0, this.serializer.serializeKey(key), value == null ? null : this.serializer.serializeValue( value)); } catch (SerializationException e) { throw new StoreException("Error serializing schema while creating the Kafka produce " + "record", e); } boolean knownSuccessfulWrite = false; try { log.trace("Sending record to KafkaStore topic: " + producerRecord); Future<RecordMetadata> ack = producer.send(producerRecord); RecordMetadata recordMetadata = ack.get(timeout, TimeUnit.MILLISECONDS); log.trace("Waiting for the local store to catch up to offset " + recordMetadata.offset()); this.lastWrittenOffset = recordMetadata.offset(); if (key instanceof SubjectKey) { setLastOffset(((SubjectKey) key).getSubject(), recordMetadata.offset()); } waitUntilKafkaReaderReachesOffset(recordMetadata.offset(), timeout); knownSuccessfulWrite = true; } catch (InterruptedException e) { throw new StoreException("Put operation interrupted while waiting for an ack from Kafka", e); } catch (ExecutionException e) { throw new StoreException("Put operation failed while waiting for an ack from Kafka", e); } catch (TimeoutException e) { throw new StoreTimeoutException( "Put operation timed out while waiting for an ack from Kafka", e); } catch (KafkaException ke) { throw new StoreException("Put operation to Kafka failed", ke); } finally { if (!knownSuccessfulWrite) { markLastWrittenOffsetInvalid(); } } return oldValue; }
KafkaStore implements Store<K, V> { @Override public V put(K key, V value) throws StoreTimeoutException, StoreException { assertInitialized(); if (key == null) { throw new StoreException("Key should not be null"); } V oldValue = get(key); ProducerRecord<byte[], byte[]> producerRecord = null; try { producerRecord = new ProducerRecord<byte[], byte[]>(topic, 0, this.serializer.serializeKey(key), value == null ? null : this.serializer.serializeValue( value)); } catch (SerializationException e) { throw new StoreException("Error serializing schema while creating the Kafka produce " + "record", e); } boolean knownSuccessfulWrite = false; try { log.trace("Sending record to KafkaStore topic: " + producerRecord); Future<RecordMetadata> ack = producer.send(producerRecord); RecordMetadata recordMetadata = ack.get(timeout, TimeUnit.MILLISECONDS); log.trace("Waiting for the local store to catch up to offset " + recordMetadata.offset()); this.lastWrittenOffset = recordMetadata.offset(); if (key instanceof SubjectKey) { setLastOffset(((SubjectKey) key).getSubject(), recordMetadata.offset()); } waitUntilKafkaReaderReachesOffset(recordMetadata.offset(), timeout); knownSuccessfulWrite = true; } catch (InterruptedException e) { throw new StoreException("Put operation interrupted while waiting for an ack from Kafka", e); } catch (ExecutionException e) { throw new StoreException("Put operation failed while waiting for an ack from Kafka", e); } catch (TimeoutException e) { throw new StoreTimeoutException( "Put operation timed out while waiting for an ack from Kafka", e); } catch (KafkaException ke) { throw new StoreException("Put operation to Kafka failed", ke); } finally { if (!knownSuccessfulWrite) { markLastWrittenOffsetInvalid(); } } return oldValue; } }
KafkaStore implements Store<K, V> { @Override public V put(K key, V value) throws StoreTimeoutException, StoreException { assertInitialized(); if (key == null) { throw new StoreException("Key should not be null"); } V oldValue = get(key); ProducerRecord<byte[], byte[]> producerRecord = null; try { producerRecord = new ProducerRecord<byte[], byte[]>(topic, 0, this.serializer.serializeKey(key), value == null ? null : this.serializer.serializeValue( value)); } catch (SerializationException e) { throw new StoreException("Error serializing schema while creating the Kafka produce " + "record", e); } boolean knownSuccessfulWrite = false; try { log.trace("Sending record to KafkaStore topic: " + producerRecord); Future<RecordMetadata> ack = producer.send(producerRecord); RecordMetadata recordMetadata = ack.get(timeout, TimeUnit.MILLISECONDS); log.trace("Waiting for the local store to catch up to offset " + recordMetadata.offset()); this.lastWrittenOffset = recordMetadata.offset(); if (key instanceof SubjectKey) { setLastOffset(((SubjectKey) key).getSubject(), recordMetadata.offset()); } waitUntilKafkaReaderReachesOffset(recordMetadata.offset(), timeout); knownSuccessfulWrite = true; } catch (InterruptedException e) { throw new StoreException("Put operation interrupted while waiting for an ack from Kafka", e); } catch (ExecutionException e) { throw new StoreException("Put operation failed while waiting for an ack from Kafka", e); } catch (TimeoutException e) { throw new StoreTimeoutException( "Put operation timed out while waiting for an ack from Kafka", e); } catch (KafkaException ke) { throw new StoreException("Put operation to Kafka failed", ke); } finally { if (!knownSuccessfulWrite) { markLastWrittenOffsetInvalid(); } } return oldValue; } KafkaStore(SchemaRegistryConfig config,
                    StoreUpdateHandler<K, V> storeUpdateHandler,
                    Serializer<K, V> serializer,
                    Store<K, V> localStore,
                    K noopKey); }
KafkaStore implements Store<K, V> { @Override public V put(K key, V value) throws StoreTimeoutException, StoreException { assertInitialized(); if (key == null) { throw new StoreException("Key should not be null"); } V oldValue = get(key); ProducerRecord<byte[], byte[]> producerRecord = null; try { producerRecord = new ProducerRecord<byte[], byte[]>(topic, 0, this.serializer.serializeKey(key), value == null ? null : this.serializer.serializeValue( value)); } catch (SerializationException e) { throw new StoreException("Error serializing schema while creating the Kafka produce " + "record", e); } boolean knownSuccessfulWrite = false; try { log.trace("Sending record to KafkaStore topic: " + producerRecord); Future<RecordMetadata> ack = producer.send(producerRecord); RecordMetadata recordMetadata = ack.get(timeout, TimeUnit.MILLISECONDS); log.trace("Waiting for the local store to catch up to offset " + recordMetadata.offset()); this.lastWrittenOffset = recordMetadata.offset(); if (key instanceof SubjectKey) { setLastOffset(((SubjectKey) key).getSubject(), recordMetadata.offset()); } waitUntilKafkaReaderReachesOffset(recordMetadata.offset(), timeout); knownSuccessfulWrite = true; } catch (InterruptedException e) { throw new StoreException("Put operation interrupted while waiting for an ack from Kafka", e); } catch (ExecutionException e) { throw new StoreException("Put operation failed while waiting for an ack from Kafka", e); } catch (TimeoutException e) { throw new StoreTimeoutException( "Put operation timed out while waiting for an ack from Kafka", e); } catch (KafkaException ke) { throw new StoreException("Put operation to Kafka failed", ke); } finally { if (!knownSuccessfulWrite) { markLastWrittenOffsetInvalid(); } } return oldValue; } KafkaStore(SchemaRegistryConfig config,
                    StoreUpdateHandler<K, V> storeUpdateHandler,
                    Serializer<K, V> serializer,
                    Store<K, V> localStore,
                    K noopKey); @Override void init(); static void addSchemaRegistryConfigsToClientProperties(SchemaRegistryConfig config,
                                                                Properties props); void waitUntilKafkaReaderReachesLastOffset(int timeoutMs); void waitUntilKafkaReaderReachesLastOffset(String subject, int timeoutMs); void markLastWrittenOffsetInvalid(); @Override V get(K key); @Override V put(K key, V value); @Override CloseableIterator<V> getAll(K key1, K key2); @Override void putAll(Map<K, V> entries); @Override V delete(K key); @Override CloseableIterator<K> getAllKeys(); @Override void flush(); @Override void close(); void waitForInit(); boolean initialized(); long lastOffset(String subject); void setLastOffset(String subject, long lastOffset); Lock leaderLock(); Lock lockFor(String subject); }
KafkaStore implements Store<K, V> { @Override public V put(K key, V value) throws StoreTimeoutException, StoreException { assertInitialized(); if (key == null) { throw new StoreException("Key should not be null"); } V oldValue = get(key); ProducerRecord<byte[], byte[]> producerRecord = null; try { producerRecord = new ProducerRecord<byte[], byte[]>(topic, 0, this.serializer.serializeKey(key), value == null ? null : this.serializer.serializeValue( value)); } catch (SerializationException e) { throw new StoreException("Error serializing schema while creating the Kafka produce " + "record", e); } boolean knownSuccessfulWrite = false; try { log.trace("Sending record to KafkaStore topic: " + producerRecord); Future<RecordMetadata> ack = producer.send(producerRecord); RecordMetadata recordMetadata = ack.get(timeout, TimeUnit.MILLISECONDS); log.trace("Waiting for the local store to catch up to offset " + recordMetadata.offset()); this.lastWrittenOffset = recordMetadata.offset(); if (key instanceof SubjectKey) { setLastOffset(((SubjectKey) key).getSubject(), recordMetadata.offset()); } waitUntilKafkaReaderReachesOffset(recordMetadata.offset(), timeout); knownSuccessfulWrite = true; } catch (InterruptedException e) { throw new StoreException("Put operation interrupted while waiting for an ack from Kafka", e); } catch (ExecutionException e) { throw new StoreException("Put operation failed while waiting for an ack from Kafka", e); } catch (TimeoutException e) { throw new StoreTimeoutException( "Put operation timed out while waiting for an ack from Kafka", e); } catch (KafkaException ke) { throw new StoreException("Put operation to Kafka failed", ke); } finally { if (!knownSuccessfulWrite) { markLastWrittenOffsetInvalid(); } } return oldValue; } KafkaStore(SchemaRegistryConfig config,
                    StoreUpdateHandler<K, V> storeUpdateHandler,
                    Serializer<K, V> serializer,
                    Store<K, V> localStore,
                    K noopKey); @Override void init(); static void addSchemaRegistryConfigsToClientProperties(SchemaRegistryConfig config,
                                                                Properties props); void waitUntilKafkaReaderReachesLastOffset(int timeoutMs); void waitUntilKafkaReaderReachesLastOffset(String subject, int timeoutMs); void markLastWrittenOffsetInvalid(); @Override V get(K key); @Override V put(K key, V value); @Override CloseableIterator<V> getAll(K key1, K key2); @Override void putAll(Map<K, V> entries); @Override V delete(K key); @Override CloseableIterator<K> getAllKeys(); @Override void flush(); @Override void close(); void waitForInit(); boolean initialized(); long lastOffset(String subject); void setLastOffset(String subject, long lastOffset); Lock leaderLock(); Lock lockFor(String subject);  }