@Test public void testWaitUntilOffset() throws Exception { String schema = TestUtils.getRandomCanonicalAvroString(1).get(0); int id1 = restApp.restClient.registerSchema(schema, "subject1"); KafkaSchemaRegistry sr = (KafkaSchemaRegistry) restApp.schemaRegistry(); KafkaStoreReaderThread readerThread = sr.getKafkaStore().getKafkaStoreReaderThread(); try { readerThread.waitUntilOffset(50L, 500L, TimeUnit.MILLISECONDS); fail("Should have timed out waiting to reach non-existent offset."); } catch (StoreTimeoutException e) { } try { readerThread.waitUntilOffset(0L, 5000L, TimeUnit.MILLISECONDS); } catch (StoreTimeoutException e) { fail("5 seconds should be more than enough time to reach offset 0 in the log."); } }
public void waitUntilOffset(long offset, long timeout, TimeUnit timeUnit) throws StoreException { if (offset < 0) { throw new StoreException("KafkaStoreReaderThread can't wait for a negative offset."); } log.trace("Waiting to read offset {}. Currently at offset {}", offset, offsetInSchemasTopic); try { offsetUpdateLock.lock(); long timeoutNs = TimeUnit.NANOSECONDS.convert(timeout, timeUnit); while ((offsetInSchemasTopic < offset) && (timeoutNs > 0)) { try { timeoutNs = offsetReachedThreshold.awaitNanos(timeoutNs); } catch (InterruptedException e) { log.debug("Interrupted while waiting for the background store reader thread to reach" + " the specified offset: " + offset, e); } } } finally { offsetUpdateLock.unlock(); } if (offsetInSchemasTopic < offset) { throw new StoreTimeoutException( "KafkaStoreReaderThread failed to reach target offset within the timeout interval. " + "targetOffset: " + offset + ", offsetReached: " + offsetInSchemasTopic + ", timeout(ms): " + TimeUnit.MILLISECONDS.convert(timeout, timeUnit)); } }
KafkaStoreReaderThread extends ShutdownableThread { public void waitUntilOffset(long offset, long timeout, TimeUnit timeUnit) throws StoreException { if (offset < 0) { throw new StoreException("KafkaStoreReaderThread can't wait for a negative offset."); } log.trace("Waiting to read offset {}. Currently at offset {}", offset, offsetInSchemasTopic); try { offsetUpdateLock.lock(); long timeoutNs = TimeUnit.NANOSECONDS.convert(timeout, timeUnit); while ((offsetInSchemasTopic < offset) && (timeoutNs > 0)) { try { timeoutNs = offsetReachedThreshold.awaitNanos(timeoutNs); } catch (InterruptedException e) { log.debug("Interrupted while waiting for the background store reader thread to reach" + " the specified offset: " + offset, e); } } } finally { offsetUpdateLock.unlock(); } if (offsetInSchemasTopic < offset) { throw new StoreTimeoutException( "KafkaStoreReaderThread failed to reach target offset within the timeout interval. " + "targetOffset: " + offset + ", offsetReached: " + offsetInSchemasTopic + ", timeout(ms): " + TimeUnit.MILLISECONDS.convert(timeout, timeUnit)); } } }
KafkaStoreReaderThread extends ShutdownableThread { public void waitUntilOffset(long offset, long timeout, TimeUnit timeUnit) throws StoreException { if (offset < 0) { throw new StoreException("KafkaStoreReaderThread can't wait for a negative offset."); } log.trace("Waiting to read offset {}. Currently at offset {}", offset, offsetInSchemasTopic); try { offsetUpdateLock.lock(); long timeoutNs = TimeUnit.NANOSECONDS.convert(timeout, timeUnit); while ((offsetInSchemasTopic < offset) && (timeoutNs > 0)) { try { timeoutNs = offsetReachedThreshold.awaitNanos(timeoutNs); } catch (InterruptedException e) { log.debug("Interrupted while waiting for the background store reader thread to reach" + " the specified offset: " + offset, e); } } } finally { offsetUpdateLock.unlock(); } if (offsetInSchemasTopic < offset) { throw new StoreTimeoutException( "KafkaStoreReaderThread failed to reach target offset within the timeout interval. " + "targetOffset: " + offset + ", offsetReached: " + offsetInSchemasTopic + ", timeout(ms): " + TimeUnit.MILLISECONDS.convert(timeout, timeUnit)); } } KafkaStoreReaderThread(String bootstrapBrokers,
                                String topic,
                                String groupId,
                                StoreUpdateHandler<K, V> storeUpdateHandler,
                                Serializer<K, V> serializer,
                                Store<K, V> localStore,
                                Producer<byte[], byte[]> producer,
                                K noopKey,
                                AtomicBoolean initialized,
                                SchemaRegistryConfig config); }
KafkaStoreReaderThread extends ShutdownableThread { public void waitUntilOffset(long offset, long timeout, TimeUnit timeUnit) throws StoreException { if (offset < 0) { throw new StoreException("KafkaStoreReaderThread can't wait for a negative offset."); } log.trace("Waiting to read offset {}. Currently at offset {}", offset, offsetInSchemasTopic); try { offsetUpdateLock.lock(); long timeoutNs = TimeUnit.NANOSECONDS.convert(timeout, timeUnit); while ((offsetInSchemasTopic < offset) && (timeoutNs > 0)) { try { timeoutNs = offsetReachedThreshold.awaitNanos(timeoutNs); } catch (InterruptedException e) { log.debug("Interrupted while waiting for the background store reader thread to reach" + " the specified offset: " + offset, e); } } } finally { offsetUpdateLock.unlock(); } if (offsetInSchemasTopic < offset) { throw new StoreTimeoutException( "KafkaStoreReaderThread failed to reach target offset within the timeout interval. " + "targetOffset: " + offset + ", offsetReached: " + offsetInSchemasTopic + ", timeout(ms): " + TimeUnit.MILLISECONDS.convert(timeout, timeUnit)); } } KafkaStoreReaderThread(String bootstrapBrokers,
                                String topic,
                                String groupId,
                                StoreUpdateHandler<K, V> storeUpdateHandler,
                                Serializer<K, V> serializer,
                                Store<K, V> localStore,
                                Producer<byte[], byte[]> producer,
                                K noopKey,
                                AtomicBoolean initialized,
                                SchemaRegistryConfig config); @Override void doWork(); @Override void shutdown(); void waitUntilOffset(long offset, long timeout, TimeUnit timeUnit); String getConsumerProperty(String key); }
KafkaStoreReaderThread extends ShutdownableThread { public void waitUntilOffset(long offset, long timeout, TimeUnit timeUnit) throws StoreException { if (offset < 0) { throw new StoreException("KafkaStoreReaderThread can't wait for a negative offset."); } log.trace("Waiting to read offset {}. Currently at offset {}", offset, offsetInSchemasTopic); try { offsetUpdateLock.lock(); long timeoutNs = TimeUnit.NANOSECONDS.convert(timeout, timeUnit); while ((offsetInSchemasTopic < offset) && (timeoutNs > 0)) { try { timeoutNs = offsetReachedThreshold.awaitNanos(timeoutNs); } catch (InterruptedException e) { log.debug("Interrupted while waiting for the background store reader thread to reach" + " the specified offset: " + offset, e); } } } finally { offsetUpdateLock.unlock(); } if (offsetInSchemasTopic < offset) { throw new StoreTimeoutException( "KafkaStoreReaderThread failed to reach target offset within the timeout interval. " + "targetOffset: " + offset + ", offsetReached: " + offsetInSchemasTopic + ", timeout(ms): " + TimeUnit.MILLISECONDS.convert(timeout, timeUnit)); } } KafkaStoreReaderThread(String bootstrapBrokers,
                                String topic,
                                String groupId,
                                StoreUpdateHandler<K, V> storeUpdateHandler,
                                Serializer<K, V> serializer,
                                Store<K, V> localStore,
                                Producer<byte[], byte[]> producer,
                                K noopKey,
                                AtomicBoolean initialized,
                                SchemaRegistryConfig config); @Override void doWork(); @Override void shutdown(); void waitUntilOffset(long offset, long timeout, TimeUnit timeUnit); String getConsumerProperty(String key);  }