import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException; //导入依赖的package包/类
/**
 * Register the schema with the kafka schema registry.
 * @param schemaName the name of the schema
 * @param schema the avro schema
 * @return the id for the schema returned from the schema registry.
 */
public int registerSchema(String schemaName, Schema schema) {
    int id;
    
    try {
        /*
         * "-value" appended to schema name per Kafka documentation.
         * See link: : http://docs.confluent.io/2.0.0/schema-registry/docs/api.html
         */
        id = client.register(schemaName+"-value", schema);
    } catch (RestClientException | IOException e) {
        LOG.error("registerSchema(): failed. Returning ID=0", e);
        id = 0;
    }

    LOG.debug("RegisterSchema(): Schema ID returned from registry for {} was {}", schemaName, id);
    
    return id;
}
 

import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException; //导入依赖的package包/类
private SchemaMetadata fetchSchemaMetadata(
    AbstractStreamCreateStatement abstractStreamCreateStatement,
    SchemaRegistryClient schemaRegistryClient,
    String kafkaTopicName) throws IOException, RestClientException {

  if (abstractStreamCreateStatement.getProperties().containsKey(KsqlConstants.AVRO_SCHEMA_ID)) {
    int schemaId;
    try {
      schemaId = Integer.parseInt(StringUtil.cleanQuotes(abstractStreamCreateStatement.getProperties().get(KsqlConstants.AVRO_SCHEMA_ID).toString()));
    } catch (NumberFormatException e) {
      throw new KsqlException(String.format("Invalid schema id property: %s.",
                                            abstractStreamCreateStatement.getProperties().get(KsqlConstants.AVRO_SCHEMA_ID).toString()));
    }
    return schemaRegistryClient.getSchemaMetadata(
        kafkaTopicName +KsqlConstants.SCHEMA_REGISTRY_VALUE_SUFFIX, schemaId);
  } else {
    return schemaRegistryClient.getLatestSchemaMetadata(kafkaTopicName +
                                                                  KsqlConstants.SCHEMA_REGISTRY_VALUE_SUFFIX);
  }
}
 

import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException; //导入依赖的package包/类
@Test
public void shouldValidatePersistentQueryResultCorrectly()
    throws IOException, RestClientException {
  SchemaRegistryClient schemaRegistryClient = mock(SchemaRegistryClient.class);
  KsqlTopic resultTopic = new KsqlTopic("testTopic", "testTopic", new KsqlAvroTopicSerDe());
  Schema resultSchema = SerDeUtil.getSchemaFromAvro(ordersAveroSchemaStr);
  PersistentQueryMetadata persistentQueryMetadata = new PersistentQueryMetadata("",
                                                                                null,
                                                                                null,
                                                                                "",
                                                                                null,
                                                                                DataSource.DataSourceType.KSTREAM,
                                                                                "",
                                                                                mock(KafkaTopicClient.class),
      resultSchema,
                                                                                resultTopic,
                                                                                null);
  org.apache.avro.Schema.Parser parser = new org.apache.avro.Schema.Parser();
  org.apache.avro.Schema avroSchema = parser.parse(ordersAveroSchemaStr);
  expect(schemaRegistryClient.testCompatibility(anyString(), EasyMock.isA(avroSchema.getClass())))
      .andReturn(true);
  replay(schemaRegistryClient);
  avroUtil.validatePersistentQueryResults(persistentQueryMetadata, schemaRegistryClient);
}
 

import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException; //导入依赖的package包/类
private void registerSchema(SchemaRegistryClient schemaRegistryClient)
    throws IOException, RestClientException {
  String ordersAveroSchemaStr = "{"
                                + "\"namespace\": \"kql\","
                                + " \"name\": \"orders\","
                                + " \"type\": \"record\","
                                + " \"fields\": ["
                                + "     {\"name\": \"ordertime\", \"type\": \"long\"},"
                                + "     {\"name\": \"orderid\",  \"type\": \"long\"},"
                                + "     {\"name\": \"itemid\", \"type\": \"string\"},"
                                + "     {\"name\": \"orderunits\", \"type\": \"double\"},"
                                + "     {\"name\": \"arraycol\", \"type\": {\"type\": \"array\", \"items\": \"double\"}},"
                                + "     {\"name\": \"mapcol\", \"type\": {\"type\": \"map\", \"values\": \"double\"}}"
                                + " ]"
                                + "}";
  org.apache.avro.Schema.Parser parser = new org.apache.avro.Schema.Parser();
  org.apache.avro.Schema avroSchema = parser.parse(ordersAveroSchemaStr);
  schemaRegistryClient.register("orders-topic" + KsqlConstants.SCHEMA_REGISTRY_VALUE_SUFFIX,
                                avroSchema);

}
 

import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException; //导入依赖的package包/类
@Override
public Schema retrieveSchema(TableId table, String topic) {
  try {
    String subject = getSubject(topic);
    logger.debug("Retrieving schema information for topic {} with subject {}", topic, subject);
    SchemaMetadata latestSchemaMetadata = schemaRegistryClient.getLatestSchemaMetadata(subject);
    org.apache.avro.Schema avroSchema = new Parser().parse(latestSchemaMetadata.getSchema());
    return avroData.toConnectSchema(avroSchema);
  } catch (IOException | RestClientException exception) {
    throw new ConnectException(
        "Exception encountered while trying to fetch latest schema metadata from Schema Registry",
        exception
    );
  }
}
 

import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException; //导入依赖的package包/类
@Test
public void shouldFailForInvalidResultAvroSchema()
    throws IOException, RestClientException {
  SchemaRegistryClient schemaRegistryClient = mock(SchemaRegistryClient.class);
  KsqlTopic resultTopic = new KsqlTopic("testTopic", "testTopic", new KsqlAvroTopicSerDe
      ());
  Schema resultSchema = SerDeUtil.getSchemaFromAvro(ordersAveroSchemaStr);
  PersistentQueryMetadata persistentQueryMetadata = new PersistentQueryMetadata("",
                                                                                null,
                                                                                null,
                                                                                "",
                                                                                null,
                                                                                DataSource.DataSourceType.KSTREAM,
                                                                                "",
                                                                                mock(KafkaTopicClient.class),
      resultSchema,
                                                                                resultTopic,
                                                                                null);
  expect(schemaRegistryClient.testCompatibility(anyString(), anyObject())).andReturn(false);
  replay(schemaRegistryClient);
  try {
    avroUtil.validatePersistentQueryResults(persistentQueryMetadata, schemaRegistryClient);
    fail();
  } catch (Exception e) {
    assertThat("Incorrect exception message", "Cannot register avro schema for testTopic since "
                                              + "it is not valid for schema registry.", equalTo(e.getMessage()));
  }
}
 

import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException; //导入依赖的package包/类
@Before
public void setUp() throws IOException, RestClientException {
  schemaRegistryClient = new MockSchemaRegistryClient();
  registerSchema(schemaRegistryClient);
  ksqlRestConfig = new KsqlRestConfig(TestKsqlResourceUtil.getDefaultKsqlConfig());
  KsqlConfig ksqlConfig = new KsqlConfig(ksqlRestConfig.getKsqlStreamsProperties());
  ksqlEngine = new KsqlEngine(ksqlConfig, new MockKafkaTopicClient(), schemaRegistryClient);
}
 

import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException; //导入依赖的package包/类
/**
 * Loads and parses a schema for the specified subject from the schema registry
 * @param subject subject for which to fetch the latest version of a schema.
 * @return parsed avro schema
 * @throws SchemaRegistryException if there is an error with the registry client
 */
public Schema loadFromRegistry(String subject) throws SchemaRegistryException {
  try {
    SchemaMetadata metadata = registryClient.getLatestSchemaMetadata(subject);
    return registryClient.getByID(metadata.getId());
  } catch (IOException | RestClientException e) {
    throw new SchemaRegistryException(e);
  }
}
 

import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException; //导入依赖的package包/类
/**
 * Looks up schema id for the specified subject from the schema registry
 * @param subject subject for which schema Id must be looked up.
 * @return the schema id
 * @throws SchemaRegistryException if there is an error with the registry client
 */
public int getSchemaIdFromSubject(String subject) throws SchemaRegistryException {
  try {
    SchemaMetadata metadata = registryClient.getLatestSchemaMetadata(subject);
    return metadata.getId();
  } catch (IOException | RestClientException e) {
    throw new SchemaRegistryException(e);
  }
}
 

import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException; //导入依赖的package包/类
@Override
protected Schema fetchSchemaByKey(Integer key) throws SchemaRegistryException {
  try {
    return this.schemaRegistryClient.getByID(key);
  } catch (IOException | RestClientException e) {
    throw new SchemaRegistryException(e);
  }
}
 

import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException; //导入依赖的package包/类
@Override
public Schema getLatestSchemaByTopic(String topic) throws SchemaRegistryException {
  String schemaName = topic + this.schemaNameSuffix;
  try {
    return new Schema.Parser().parse(this.schemaRegistryClient.getLatestSchemaMetadata(schemaName).getSchema());
  } catch (IOException | RestClientException e) {
    log.error("Failed to get schema for topic " + topic + "; subject " + schemaName);
    throw new SchemaRegistryException(e);
  }
}
 

import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException; //导入依赖的package包/类
@Override
public Integer register(Schema schema, String name) throws SchemaRegistryException {
  try {
    String schemaName = name + this.schemaNameSuffix;
    return this.schemaRegistryClient.register(schemaName, schema);
  } catch (IOException | RestClientException e) {
    throw new SchemaRegistryException(e);
  }
}
 

import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException; //导入依赖的package包/类
@Test
public void testConfluentAvroDeserializer() throws IOException, RestClientException {
  WorkUnitState mockWorkUnitState = getMockWorkUnitState();
  mockWorkUnitState.setProp("schema.registry.url", TEST_URL);

  Schema schema = SchemaBuilder.record(TEST_RECORD_NAME)
      .namespace(TEST_NAMESPACE).fields()
      .name(TEST_FIELD_NAME).type().stringType().noDefault()
      .endRecord();

  GenericRecord testGenericRecord = new GenericRecordBuilder(schema).set(TEST_FIELD_NAME, "testValue").build();

  SchemaRegistryClient mockSchemaRegistryClient = mock(SchemaRegistryClient.class);
  when(mockSchemaRegistryClient.getByID(any(Integer.class))).thenReturn(schema);

  Serializer<Object> kafkaEncoder = new KafkaAvroSerializer(mockSchemaRegistryClient);
  Deserializer<Object> kafkaDecoder = new KafkaAvroDeserializer(mockSchemaRegistryClient);

  ByteBuffer testGenericRecordByteBuffer =
      ByteBuffer.wrap(kafkaEncoder.serialize(TEST_TOPIC_NAME, testGenericRecord));

  KafkaSchemaRegistry<Integer, Schema> mockKafkaSchemaRegistry = mock(KafkaSchemaRegistry.class);
  KafkaDeserializerExtractor kafkaDecoderExtractor =
      new KafkaDeserializerExtractor(mockWorkUnitState,
          Optional.fromNullable(Deserializers.CONFLUENT_AVRO), kafkaDecoder, mockKafkaSchemaRegistry);

  ByteArrayBasedKafkaRecord mockMessageAndOffset = getMockMessageAndOffset(testGenericRecordByteBuffer);

  Assert.assertEquals(kafkaDecoderExtractor.decodeRecord(mockMessageAndOffset), testGenericRecord);
}
 

import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException; //导入依赖的package包/类
@Test
public void testConfluentAvroDeserializerForSchemaEvolution() throws IOException, RestClientException, SchemaRegistryException {
  WorkUnitState mockWorkUnitState = getMockWorkUnitState();
  mockWorkUnitState.setProp("schema.registry.url", TEST_URL);

  Schema schemaV1 = SchemaBuilder.record(TEST_RECORD_NAME)
      .namespace(TEST_NAMESPACE).fields()
      .name(TEST_FIELD_NAME).type().stringType().noDefault()
      .endRecord();

  Schema schemaV2 = SchemaBuilder.record(TEST_RECORD_NAME)
      .namespace(TEST_NAMESPACE).fields()
      .name(TEST_FIELD_NAME).type().stringType().noDefault()
      .optionalString(TEST_FIELD_NAME2).endRecord();

  GenericRecord testGenericRecord = new GenericRecordBuilder(schemaV1).set(TEST_FIELD_NAME, "testValue").build();

  SchemaRegistryClient mockSchemaRegistryClient = mock(SchemaRegistryClient.class);
  when(mockSchemaRegistryClient.getByID(any(Integer.class))).thenReturn(schemaV1);

  Serializer<Object> kafkaEncoder = new KafkaAvroSerializer(mockSchemaRegistryClient);
  Deserializer<Object> kafkaDecoder = new KafkaAvroDeserializer(mockSchemaRegistryClient);

  ByteBuffer testGenericRecordByteBuffer =
      ByteBuffer.wrap(kafkaEncoder.serialize(TEST_TOPIC_NAME, testGenericRecord));

  KafkaSchemaRegistry<Integer, Schema> mockKafkaSchemaRegistry = mock(KafkaSchemaRegistry.class);
  when(mockKafkaSchemaRegistry.getLatestSchemaByTopic(TEST_TOPIC_NAME)).thenReturn(schemaV2);

  KafkaDeserializerExtractor kafkaDecoderExtractor = new KafkaDeserializerExtractor(mockWorkUnitState,
      Optional.fromNullable(Deserializers.CONFLUENT_AVRO), kafkaDecoder, mockKafkaSchemaRegistry);
  when(kafkaDecoderExtractor.getSchema()).thenReturn(schemaV2);

  ByteArrayBasedKafkaRecord mockMessageAndOffset = getMockMessageAndOffset(testGenericRecordByteBuffer);

  GenericRecord received = (GenericRecord) kafkaDecoderExtractor.decodeRecord(mockMessageAndOffset);
  Assert.assertEquals(received.toString(), "{\"testField\": \"testValue\", \"testField2\": null}");

}