@Test public void testConfluentShouldNotQuerySchemaRegistryWhenTheGapIsZero() throws IOException, RestClientException, SchemaRegistryException { WorkUnitState mockWorkUnitState = getMockWorkUnitState(0L, 0L); mockWorkUnitState.setProp("schema.registry.url", TEST_URL); SchemaRegistryClient mockSchemaRegistryClient = mock(SchemaRegistryClient.class); Deserializer<Object> kafkaDecoder = new KafkaAvroDeserializer(mockSchemaRegistryClient); KafkaSchemaRegistry<Integer, Schema> mockKafkaSchemaRegistry = mock(KafkaSchemaRegistry.class); KafkaDeserializerExtractor kafkaDecoderExtractor = new KafkaDeserializerExtractor(mockWorkUnitState, Optional.fromNullable(Deserializers.CONFLUENT_AVRO), kafkaDecoder, mockKafkaSchemaRegistry); verify(mockKafkaSchemaRegistry, never()).getLatestSchemaByTopic(any()); kafkaDecoderExtractor.getSchema(); }
@Override public Object getSchema() { try { LOG.info("Getting schema for {}. Gap: {} HighWaterMark: {}", this.topicName, this.lowWatermark.getGap(this.highWatermark)); if (this.lowWatermark.getGap(this.highWatermark) == 0) { LOG.info("Not getting schema for {} as the gap between high and low watermark is 0", this.topicName); return null; } return this.kafkaSchemaRegistry.getLatestSchemaByTopic(this.topicName); } catch (SchemaRegistryException e) { throw new RuntimeException(e); } }
KafkaDeserializerExtractor extends KafkaExtractor<Object, Object> { @Override public Object getSchema() { try { LOG.info("Getting schema for {}. Gap: {} HighWaterMark: {}", this.topicName, this.lowWatermark.getGap(this.highWatermark)); if (this.lowWatermark.getGap(this.highWatermark) == 0) { LOG.info("Not getting schema for {} as the gap between high and low watermark is 0", this.topicName); return null; } return this.kafkaSchemaRegistry.getLatestSchemaByTopic(this.topicName); } catch (SchemaRegistryException e) { throw new RuntimeException(e); } } }
KafkaDeserializerExtractor extends KafkaExtractor<Object, Object> { @Override public Object getSchema() { try { LOG.info("Getting schema for {}. Gap: {} HighWaterMark: {}", this.topicName, this.lowWatermark.getGap(this.highWatermark)); if (this.lowWatermark.getGap(this.highWatermark) == 0) { LOG.info("Not getting schema for {} as the gap between high and low watermark is 0", this.topicName); return null; } return this.kafkaSchemaRegistry.getLatestSchemaByTopic(this.topicName); } catch (SchemaRegistryException e) { throw new RuntimeException(e); } } KafkaDeserializerExtractor(WorkUnitState state); private  KafkaDeserializerExtractor(WorkUnitState state, Optional<Deserializers> deserializerType); @VisibleForTesting  KafkaDeserializerExtractor(WorkUnitState state, Optional<Deserializers> deserializerType,
      Deserializer<?> kafkaDeserializer, KafkaSchemaRegistry<?, ?> kafkaSchemaRegistry); }
KafkaDeserializerExtractor extends KafkaExtractor<Object, Object> { @Override public Object getSchema() { try { LOG.info("Getting schema for {}. Gap: {} HighWaterMark: {}", this.topicName, this.lowWatermark.getGap(this.highWatermark)); if (this.lowWatermark.getGap(this.highWatermark) == 0) { LOG.info("Not getting schema for {} as the gap between high and low watermark is 0", this.topicName); return null; } return this.kafkaSchemaRegistry.getLatestSchemaByTopic(this.topicName); } catch (SchemaRegistryException e) { throw new RuntimeException(e); } } KafkaDeserializerExtractor(WorkUnitState state); private  KafkaDeserializerExtractor(WorkUnitState state, Optional<Deserializers> deserializerType); @VisibleForTesting  KafkaDeserializerExtractor(WorkUnitState state, Optional<Deserializers> deserializerType,
      Deserializer<?> kafkaDeserializer, KafkaSchemaRegistry<?, ?> kafkaSchemaRegistry); @Override Object getSchema(); }
KafkaDeserializerExtractor extends KafkaExtractor<Object, Object> { @Override public Object getSchema() { try { LOG.info("Getting schema for {}. Gap: {} HighWaterMark: {}", this.topicName, this.lowWatermark.getGap(this.highWatermark)); if (this.lowWatermark.getGap(this.highWatermark) == 0) { LOG.info("Not getting schema for {} as the gap between high and low watermark is 0", this.topicName); return null; } return this.kafkaSchemaRegistry.getLatestSchemaByTopic(this.topicName); } catch (SchemaRegistryException e) { throw new RuntimeException(e); } } KafkaDeserializerExtractor(WorkUnitState state); private  KafkaDeserializerExtractor(WorkUnitState state, Optional<Deserializers> deserializerType); @VisibleForTesting  KafkaDeserializerExtractor(WorkUnitState state, Optional<Deserializers> deserializerType,
      Deserializer<?> kafkaDeserializer, KafkaSchemaRegistry<?, ?> kafkaSchemaRegistry); @Override Object getSchema(); static final String KAFKA_DESERIALIZER_TYPE; }