@Override
  public SchemaPath getSchemaPath(TopicPath topicPath) throws IOException {
    Topic topic = pubsub.projects().topics().get(topicPath.getPath()).execute();
    if (topic.getSchemaSettings() == null) {
      return null;
    }
    String schemaPath = topic.getSchemaSettings().getSchema();
    if (schemaPath.equals(SchemaPath.DELETED_SCHEMA_PATH)) {
      return null;
    }
    return PubsubClient.schemaPathFromPath(schemaPath);
  }
@Test
  public void testGetSchemaPath() throws IOException {
    TopicPath topicDoesNotExist =
        PubsubClient.topicPathFromPath("projects/testProject/topics/idontexist");
    TopicPath topicExistsDeletedSchema =
        PubsubClient.topicPathFromPath("projects/testProject/topics/deletedSchema");
    TopicPath topicExistsNoSchema =
        PubsubClient.topicPathFromPath("projects/testProject/topics/noSchema");
    TopicPath topicExistsSchema =
        PubsubClient.topicPathFromPath("projects/testProject/topics/topicWithSchema");
    when(mockPubsub.projects().topics().get(topicDoesNotExist.getPath()).execute())
        .thenThrow(
            new IOException(
                String.format("topic does not exist: %s", topicDoesNotExist.getPath())));
    when(mockPubsub.projects().topics().get(topicExistsDeletedSchema.getPath()).execute())
        .thenReturn(
            new Topic()
                .setName(topicExistsDeletedSchema.getName())
                .setSchemaSettings(
                    new SchemaSettings().setSchema(PubsubClient.SchemaPath.DELETED_SCHEMA_PATH)));
    when(mockPubsub.projects().topics().get(topicExistsNoSchema.getPath()).execute())
        .thenReturn(new Topic().setName(topicExistsNoSchema.getName()));
    when(mockPubsub.projects().topics().get(topicExistsSchema.getPath()).execute())
        .thenReturn(
            new Topic()
                .setName(topicExistsSchema.getName())
                .setSchemaSettings(new SchemaSettings().setSchema(SCHEMA.getPath())));

    client = new PubsubJsonClient(null, null, mockPubsub);

    assertThrows(
        "topic does not exist", IOException.class, () -> client.getSchemaPath(topicDoesNotExist));

    assertNull("schema for topic is deleted", client.getSchemaPath(topicExistsDeletedSchema));

    assertNull("topic has no schema", client.getSchemaPath(topicExistsNoSchema));

    assertEquals(SCHEMA.getPath(), client.getSchemaPath(topicExistsSchema).getPath());
  }