@Override
    protected void handleGetSchema(CommandGetSchema commandGetSchema) {
        checkArgument(state == State.Connected);
        if (log.isDebugEnabled()) {
            if (commandGetSchema.hasSchemaVersion()) {
                log.debug("Received CommandGetSchema call from {}, schemaVersion: {}, topic: {}, requestId: {}",
                        remoteAddress, new String(commandGetSchema.getSchemaVersion()),
                        commandGetSchema.getTopic(), commandGetSchema.getRequestId());
            } else {
                log.debug("Received CommandGetSchema call from {}, schemaVersion: {}, topic: {}, requestId: {}",
                        remoteAddress, null,
                        commandGetSchema.getTopic(), commandGetSchema.getRequestId());
            }
        }

        long requestId = commandGetSchema.getRequestId();
        SchemaVersion schemaVersion = SchemaVersion.Latest;
        if (commandGetSchema.hasSchemaVersion()) {
            if (commandGetSchema.getSchemaVersion().length == 0) {
                commandSender.sendGetSchemaErrorResponse(requestId, ServerError.IncompatibleSchema,
                        "Empty schema version");
                return;
            }
            schemaVersion = schemaService.versionFromBytes(commandGetSchema.getSchemaVersion());
        }

        final String topic = commandGetSchema.getTopic();
        String schemaName;
        try {
            schemaName = TopicName.get(topic).getSchemaName();
        } catch (Throwable t) {
            commandSender.sendGetSchemaErrorResponse(requestId, ServerError.InvalidTopicName, t.getMessage());
            return;
        }

        schemaService.getSchema(schemaName, schemaVersion).thenAccept(schemaAndMetadata -> {
            if (schemaAndMetadata == null) {
                commandSender.sendGetSchemaErrorResponse(requestId, ServerError.TopicNotFound,
                        String.format("Topic not found or no-schema %s", topic));
            } else {
                commandSender.sendGetSchemaResponse(requestId,
                        SchemaInfoUtil.newSchemaInfo(schemaName, schemaAndMetadata.schema), schemaAndMetadata.version);
            }
        }).exceptionally(ex -> {
            commandSender.sendGetSchemaErrorResponse(requestId, ServerError.UnknownError, ex.getMessage());
            return null;
        });
    }
@Test(expectedExceptions = IllegalArgumentException.class)
    public void shouldFailHandleGetSchema() throws Exception {
        ServerCnx serverCnx = mock(ServerCnx.class, CALLS_REAL_METHODS);
        Field stateUpdater = ServerCnx.class.getDeclaredField("state");
        stateUpdater.setAccessible(true);
        stateUpdater.set(serverCnx, ServerCnx.State.Failed);
        serverCnx.handleGetSchema(any());
    }