API First - это принцип разработки, по которому API вашего приложения является наиболее важной частью вашего приложения. Этот принцип применяется как для клиентской разработки, так и для серверной. Ранее мы уже говорили о том, как можно реализовать принципы API First для приложений, использующих http интерфейсы (API-First и микросервисы). В этот раз мы поговорим о том, как можно реализовать принципы API First, если ваши приложения используют брокеры сообщений для общения друг с другом (такие как Kafka, RabbitMq и другие). Ниже приведена типичная ошибка при рассогласовании протоколов взаимодействия.

Проблемы документирования интерфейсов при работе с брокерами сообщений

Если для документирования http интерфейсов существует спецификация OpenAPI, то для асинхронного сообщения с использованием брокеров сообщений мы не нашли подобного инструмента. На проектах выбирают, как правило, следующие способы документирования таких интерфейсов:

• файлы (doc, pdf, excel);

• Wiki.

Формат такого рода документов от проекта к проекту (или внутри одного проекта) разный. Кроме отсутствия единого формата документирования таких интерфейсов есть и другая проблема - рассинхронизация версии документации и версии приложения. На разных стендах может быть разная версия приложения, и сопоставление документации и версии приложения становится нетривиальной задачей. 

OpenAPI и брокеры сообщений

Увидев текущие проблемы документирования интерфейсов, использующих брокеры сообщений, мы подумали, как их можно решить. Для решения проблемы отсутствия единого формата документирования решили использовать OpenAPI нотацию. Преимущества OpenAPI:

• известная для аналитиков, QA инженеров и разработчиков нотация;

• существуют средства чтения и проверки нотации;

• существуют средства для создания "моков";

• существуют решения для генерации кода по спецификации. 

Как мы знаем, OpenAPI - это нотация для описания http интерфейсов. Соответственно, вся инфраструктура вокруг OpenAPI спецификации построена для http. Для использования OpenAPI инфраструктуры необходимы доработки.

API First и брокеры сообщений - обзор существующих решений

Рассмотрим текущие решения, которые уже существуют для реализации подхода API first с интерфейсами, использующие брокеры сообщений. Их достоинства и недостатки рассмотрены с точки зрения наших потребностей. 

Springwolf

URL: https://github.com/springwolf/springwolf-core

Позволяет создать спецификацию в формате AsyncAPI - формат для описания асинхронного сообщения. 

При подключении библиотеки к своему проекту возникли некоторые проблемы (так и не получилось это сделать), хотя по документации подключение библиотеки Springwolf достаточно простое. За несколько попыток подключения нашлось много неточностей в документации. Пример из репозитория запустился. В примере есть API, которые принимают Headers, но я не нашла отражение этого в  сгенерированной документации. Также, headers в сгенерированном UI никак не отражаются, и они не требуются при формировании message (можно опубликовать message без headers).

Достоинства:

• удобный UI;

• поддержка SASL авторизации к Kafka;

• простота подключения (по документации).

Недостатки:

• сгенерированный библиотекой UI невозможно объединить с Swagger UI;

• небольшое сообщество пользователей;

• совместима только с версией java не ниже 17;

• не подходящая нам работа с headers;

• ошибки в документации.

AsyncAPI Generator

URL: https://www.asyncapi.com/tools/generator

AsyncApi - протокол для описания асинхронного общения между приложениями. Протокол позволяет проектировать приложения в Event-Driven Architecture. У OpenAPI и у AsyncAPI есть много общего (например, описание моделей). На данный момент для работы c AsyncAPI (как и c OpenAPI) существует:

• online редактор https://studio.asyncapi.com/;

• offline клиент https://www.asyncapi.com/tools/cli;

• генератор кода https://www.asyncapi.com/tools/generator;

• генератор моделей https://www.asyncapi.com/tools/modelina (онлайн версия https://modelina.org/playground?language=java).

При использовании генератора кода не удалось сгенерировать только отдельную часть приложения - приложение генерируется целиком. Протокол AsyncAPI действительно очень похож на своего родителя OpenAPI (как это и заявлено в документации). Генератор моделей Modelina может принимать спецификации в формате как AsyncAPI, так и OpenAPI, но он нас не устроил. Modelina не добавляет правила валидации в сгенерированный код, даже если правила валидации указаны в спецификации. Инфраструктура вокруг протокола AsyncAPI сейчас активно развивается, и в ней происходит много изменений, но все продукты вокруг AsyncAPI написаны не на Java (кроме Springwolf), что усложняет их “подгонку” под наши нужды Java командами.

Достоинства:

• наличие богатой инфраструктуры;

• поддержка AsyncAPI - специального протокола для описания взаимодействия приложений в Event-Driven Architecture.

Недостатки:

• AsyncAPI - слишком молодой протокол, чтобы использовать его в production, он появился только в 2019-2020 году;

• небольшое сообщество пользователей;

• генерацию кода приходится делать с помощью другого приложения (нет maven или gradle плагина);

• решение написано не на Java - сложность поддержки генератора для Java команд;

• игнорирование правил валидации моделей при генерации кода;

• нет возможности сгенерировать отдельно код клиента для сервиса, использующий Kafka в качестве средства взаимодействия с ним - можно сгенерировать либо только модели, либо приложение целиком (с конфигурациями и подключением).

Оба этих решения нам не подошли, потому что: 

• не устроило качество генерации моделей - игнорируются правила валидации;

• не устроила работа с хедерами;

• текущие решения по генерации кода сложно встроить в процесс сборки приложения;

• небольшое сообщество пользователей. 

Когда мы следовали принципам API First при разработке приложения с http интерфейсом мы сделали следующие шаги (подробнее можно прочитать в статье API-First и микросервисы): 

• создали и опубликовали в mvn- репозиторий (nexus) спецификацию API как отдельный артефакт;

• при сборке приложения, реализующего API, забирали спецификацию и генерировали по ней модели и интерфейсы Spring контроллеров;

• при сборке клиента этого приложения забирали спецификацию и генерировали по ней модели и интерфейс Feign client.

После исследования уже имеющихся решений мы начали разрабатывать собственное решение для реализации подхода API First при разработке приложений, использующих брокеры сообщений в качестве средства интеграции. Мы решили дать возможность полностью повторить шаги генерации кода и создания спецификации как и в случае использования http интерфейсов.

AxenAPI - разработка Axenix

URL: https://github.com/AxenAPI

Итак, у нас появилась цель - воссоздать API First для приложений, использующих брокеры сообщений. Одним из наиболее распространенных решений для асинхронного общения между сервисами является Kafka. Решили начать с реализации этих идей именно для неё.

Для того чтобы начать использовать подход API First необходимо:

0. Инициировать спецификацию. Этот шаг необходим, если вы хотите начать использовать подход API First в уже разрабатываемом долго продукте.

1. Внедрить генерацию кода по спецификации в процесс разработки.

2. Использовать сгенерированный код.

AxenAPI состоит из двух частей:

1. AxenAPI Gradle plugin - gradle плагин, позволяющий генерировать код по спецификации интерфейса, реализованного с помощью брокера сообщений.

2. AxenAPI library - библиотека, позволяющая генерировать контроллеры для ваших kafka-listeners. Далее, с помощью уже имеющихся средств (например, https://springdoc.org/) можно сгенерировать спецификацию OpenAPI.

Основным шагом при использования подхода API First является генерация кода по спецификации, поэтому сначала поговорим о плагине, позволяющим генерировать код.

AxenAPI Gradle plugin. Генерация кода

Для того чтобы обеспечить полный цикл API First необходимо генерировать по спецификации код сервера и клиента. Это дает гарантию, что ваше приложение реализует данное в спецификации API. Рассмотрим генерацию кода сервера. За несколько шагов вы можете избавить себя от рутины и сгенерировать код моделей и обработчиков событий (handlers), а не писать их самостоятельно.

AxenAPI Gradle plugin. Подключение

1. Подключите плагин:

plugins {
  //....
    id 'axenapi-generator-plugin' version '1.0.0'
}

2. Добавьте в проект json со спецификацией. Мы добавили в ресурсы проекта (...\src\main\resources\test.json) следующий json:

{
  "openapi": "3.0.1",
  "info": {
    "title": "App API",
    "version": "snapshot"
  },
  "servers": [
    {
      "url": "http://axenapi.demo",
      "description": "Generated server url"
    }
  ],
  "paths": {
    "/kafka/example_group/example_topic/ExampleIn": {
      "post": {
        "description": "example handler",
        "operationId": "ExampleHandler",
        "tags": ["example"],
        "security": [{
          "Internal-Token": []
        }],
        "requestBody": {
          "description": "Example in",
          "content": {
            "*/*": {
              "schema": {
                "$ref": "#/components/schemas/ExampleIn"
              }
            }
          }
        },
        "responses": {
          "200": {
            "description": "Example out",
            "content": {
              "*/*": {
                "schema": {
                  "$ref": "#/components/schemas/ExampleOut"
                }
              }
            }
          }
        }
      }
    }

  },
  "components": {
    "schemas": {
      "ExampleIn": {
        "type": "object",
        "properties": {
          "message": {
            "type": "string",
            "description": "example field"
          }
        },
        "description": "example message"
      },
      "ExampleOut": {
        "type": "object",
        "properties": {
          "message": {
            "type": "string",
            "description": "example field"
          }
        },
        "description": "example message"
      }
    },
    "securitySchemes": {
      "Internal-Token": {
        "type": "apiKey",
        "name": "SERVICE_ACCESS_TOKEN",
        "in": "header"
      }
    }
  }
}

3. Добавьте настройки для плагина в build.gradle:

codegenData {
    openApiPath =  getProjectDir().getAbsolutePath() + '/src/main/resources/test.json'
    outDir = getProjectDir().getAbsolutePath() + '/build'
    srcDir = 'src/main/java'
    listenerPackage = 'axenapi.listener'
    modelPackage = 'axenapi.model'
    kafkaClient = false
}

4. Добавьте зависимость шага компиляции от шага генерации кода. Для этого вставьте в build.gradle следующие строки:

compileJava {   dependsOn "generateKafka"}

5. Соберите ваш проект - выполните gradle build.

   Вам сгенерировался код моделей в путь build/src/main/java/axenapi, и теперь вы можете подключить папку build/src в gradle.build как source:

   Давайте посмотрим на результат:

   

package axenapi.generated;

import org.axenix.axenapi.annotation.KafkaHandlerDescription;
import org.axenix.axenapi.annotation.KafkaHandlerTags;
import org.axenix.axenapi.annotation.KafkaSecured;
import org.springframework.kafka.annotation.KafkaHandler;
import org.springframework.kafka.annotation.KafkaListener;
import org.axenix.axenapi.annotation.KafkaHandlerHeader;
import org.axenix.axenapi.annotation.KafkaHandlerHeaders;
import axenapi.generated.model.ExampleIn;
import axenapi.generated.model.ExampleOut;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.messaging.handler.annotation.Headers;
import java.util.Map;

//@KafkaListener(topics = "example_topic", groupId = "example_group")
public interface ExampleTopicExampleGroupListener {

    @KafkaHandler
    @KafkaHandlerDescription("example handler")
    @KafkaHandlerTags(tags = { "example" })
    @KafkaHandlerHeaders(headers = {
    })

    ExampleOut handleExampleIn(@Payload ExampleIn examplein, @Headers Map<String, String> headers);

}

AxenAPI Gradle plugin. Параметры плагина

Сначала опишем подробнее параметры из приведенного выше примера.

codegenData {
    openApiPath =  getProjectDir().getAbsolutePath() + '/src/main/resources/test.json'
    outDir = getProjectDir().getAbsolutePath() + '/build'
    srcDir = 'src/main/java'
    listenerPackage = 'axenapi.listener'
    modelPackage = 'axenapi.model'
    kafkaClient = false
}

• openApiPath - путь, по которому лежит ваш json со спецификацией;

• outDir - каталог, в который будет помещен сгенерированный код;

• srcDir - структура каталогов для сгенерированного кода;

• listenerPackage - package, в который будут сгенерированы listeners интерфейсы;

• modelPackage - package, в который будут сгенерированы модели (Data Transfer Object);

• kafkaClient - если false, то будет сгенерирован код сервера, а если true - то код клиента.

AxenAPI Gradle plugin. Генерация клиента

Рассмотрим второй пример генерации - генерация клиента. Для этого измените значение параметра kafkaClient на true из примера выше. Соберем проект и посмотрим результат генерации:

Мы видим сгенерированный producer и его имплементацию, сервис для отправки сообщений в Kafka и его имплементацию, файлы для автоконфигурации spring-boot приложений. Ниже приведены примеры сгенерированного кода.

package axenapi.generated.impl;

import service.KafkaSenderService;
import org.springframework.stereotype.Component;
import axenapi.generated.ExampleTopicExampleGroupProducer;
import java.util.Map;
import axenapi.generated.model.ExampleIn;
import axenapi.generated.model.ExampleOut;

@Component
public class ExampleTopicExampleGroupProducerImpl implements ExampleTopicExampleGroupProducer {

    private final KafkaSenderService kafkaSenderService;

    public ExampleTopicExampleGroupProducerImpl(KafkaSenderService kafkaSenderService) {
        this.kafkaSenderService = kafkaSenderService;
    }

    @Override
    public void sendExampleIn(ExampleIn examplein, Map<String, String> params) {
        kafkaSenderService.send("example_topic", examplein, params);
    }

}

package service.impl;

import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.stereotype.Service;
import service.KafkaSenderService;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.Map;

import org.springframework.kafka.support.converter.MessagingMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;

import java.nio.charset.StandardCharsets;
import java.util.UUID;

@Service
public class KafkaSenderServiceImpl implements KafkaSenderService {

    private String messageIdName = "kafka_messageId";
    private String correlationIdName = "kafka_correlationId";
    private Boolean sendBytes = true;
    private Boolean generateMessageId = true;
    private Boolean generateCorrelationId = true;

    private final KafkaTemplate<String, Object> kafkaTemplate;
    private final MessagingMessageConverter converter;

    public KafkaSenderServiceImpl(KafkaTemplate<String, Object> kafkaTemplate,
                                  MessagingMessageConverter converter) {
        this.kafkaTemplate = kafkaTemplate;
        this.converter = converter;
    }

    public KafkaTemplate<String, Object> getKafkaTemplate() {
        return kafkaTemplate;
    }

    public void send(String topicName, Object message, Map<String, String> params) {

        MessageHeaderAccessor headerAccessor = new MessageHeaderAccessor();
        if (generateMessageId) {
            if (sendBytes) {
                headerAccessor.setHeader(messageIdName, UUID.randomUUID().toString().getBytes(StandardCharsets.UTF_8));
            } else {
                headerAccessor.setHeader(messageIdName, UUID.randomUUID());
            }
        }
        if (generateCorrelationId) {
            if (sendBytes) {
                headerAccessor.setHeader(correlationIdName, UUID.randomUUID().toString().getBytes(StandardCharsets.UTF_8));
            } else {
                headerAccessor.setHeader(correlationIdName, UUID.randomUUID());
            }
        }
        if (params != null && params.size() > 0) {
            for (var entry : params.entrySet()) {
                if (sendBytes) {
                    headerAccessor.setHeader(entry.getKey(), entry.getValue().getBytes(StandardCharsets.UTF_8));
                } else {
                    headerAccessor.setHeader(entry.getKey(), entry.getValue());
                }
            }
        }
        Message<Object> msg = MessageBuilder.createMessage(message, headerAccessor.getMessageHeaders());
        ProducerRecord producerRecord = converter.fromMessage(msg, topicName);
        kafkaTemplate.send(producerRecord);
    }
}

package config;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.serializer.JsonSerializer;
import org.springframework.kafka.transaction.KafkaTransactionManager;

import java.util.HashMap;
import java.util.Map;

public class KafkaProducerConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String kafkaBootstrap;

    @Bean("producerFactory")
    @ConditionalOnMissingBean
    public ProducerFactory<String, Object> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(
                ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                kafkaBootstrap);
        configProps.put(
                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class);
        configProps.put(
                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                JsonSerializer.class);
        configProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
        configProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "prod-1");
        DefaultKafkaProducerFactory<String, Object> factory = new DefaultKafkaProducerFactory<>(configProps);
        //factory.setTransactionIdPrefix("prod");
        return factory;
    }

    @Bean
    public KafkaTransactionManager kafkaTransactionManager(ProducerFactory<String, Object> producerFactory) {
        return new KafkaTransactionManager(producerFactory);
    }

    @Bean
    @ConditionalOnMissingBean
    public KafkaTemplate<String, Object> kafkaTemplate(ProducerFactory<String, Object> producerFactory) {
        return new KafkaTemplate<>(producerFactory);
    }
}

package config;

import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.converter.MessagingMessageConverter;
import service.KafkaSenderService;
import service.impl.KafkaSenderServiceImpl;
import org.springframework.context.annotation.ComponentScan;

@ComponentScan("axenapi")
public class KafkaSenderServiceConfig {

    private final KafkaTemplate<String, Object> kafkaTemplate;

    public KafkaSenderServiceConfig(KafkaTemplate<String, Object> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    @Bean
    @ConditionalOnMissingBean
    public MessagingMessageConverter converter() {
        MessagingMessageConverter converter = new MessagingMessageConverter();
        converter.setGenerateMessageId(true);
        converter.setGenerateTimestamp(true);
        return converter;
    }

    @Bean
    @ConditionalOnMissingBean
    public KafkaSenderService kafkaSenderService(MessagingMessageConverter converter) {
        return new KafkaSenderServiceImpl(kafkaTemplate, converter);
    }
}

Мы рассмотрели примеры генерации кода по спецификации - основной шаг при использовании подхода API First. Теперь давайте рассмотрим AxenAPI library. AxenAPI library позволяет решить проблему инициирования спецификации.

AxenAPI library

AxenAPI library - библиотека, позволяющая генерировать контроллеры для ваших kafka-listeners. Далее, с помощью уже имеющихся средств (например, https://springdoc.org/) можно сгенерировать спецификацию OpenAPI. Http запрос будет перенаправлен в соответствующие топики. Библиотека поддерживает стандартные аннотации библиотеки io.swagger.core.v3 (https://github.com/swagger-api/swagger-core). Если в приложение подключен swagger-ui, то ваши listeners будут отражены в swagger-ui (т.к. для них сгенерированы контроллеры). Сгенерированную библиотекой OpenAPI документацию можно импортировать в Postman или другие средства тестирования http. 

AxenAPI library. Подключение

Приступим к подключению AxenAPI. Это делается буквально за несколько шагов:

1. Подключите нужные зависимости. Для этого необходимо прописать следующие зависимости в gradle.build:

    //	swagger for kafka
    annotationProcessor "org.axenix:axenapi:1.0.0"
    implementation ("org.axenix:axenapi:1.0.0")

2. Добавьте в зависимости аннотации Swagger (этот шаг можно пропустить, если вы подключите Swagger-UI):

implementation 'io.swagger.core.v3:swagger-annotations:2.2.10'

3. В application.yml вашего приложения добавьте:

axenapi.kafka.swagger.enabled: true

4. Соберите проект. Во время сборки проекта сгенерируются контроллеры для ваших listeners.

5. Добавьте Swager-UI в ваше приложение, чтобы увидеть результат - это самый простой способ  воспользоваться и наглядно увидеть ваше API и все используемые в нем модели:

implementation 'org.springdoc:springdoc-openapi-ui:1.6.13' 

5. Запустите приложение.

6. Затем откройте http://<host>:<port>/swagger-ui/index.html в вашем браузере, и вы увидите Swagger-UI, в котором присутствуют Post методы для отправки сообщений в соответствующий топик и группу (информация о топике и группе есть в url метода).

6. Разделите спецификацию и Swagger-UI на две части, если у вас в приложении есть и Kafka handlers и http API. Для этого добавьте следующую конфигурацию для Springdoc:

@Configuration
public class OpenApiConfiguration {
    @Bean
    GroupedOpenApi restApis() {
        return GroupedOpenApi.builder().group("kafka").pathsToMatch("/**/kafka/**").build();
    }

    @Bean
    GroupedOpenApi kafkaApis() {
        return GroupedOpenApi.builder().group("rest").pathsToMatch("/**/users/**").build();
    }
}

Как уже говорилось, во время сборки проекта будут генерироваться контроллеры для ваших listeners. Самый простой способ наглядно увидеть результат - подключить Swagger-UI. Можно сгенерировать спецификацию с помощью gradle плагинов (например, springdoc-openapi-gradle-plugin).

В application.yml (или application.properties) можно прописать следующие настройки:

• axenapi.kafka.swagger.enabled: true/false - сгенерированные контроллеры будут подняты/не подняты при старте приложения (значение по умолчанию - false);

• axenapi.headers.sendBytes: true/false - header’ы либо будут присылаться вместе с типами, либо все header’ы будут считаться массивом байт (значение по умолчанию false).

Можно добавить файл axenapi.properties в корень проекта для настройки annotation processor:

• package = com.example.demo - работа будет производиться с listeners только из указанного пакета; если не указано, то работа будет идти со всеми listeners из вашего проекта;

• kafka.handler.annotaion = ru.axteam.MyHandler - своя аннотация для поиска listeners, которую необходимо указывать, если на вашем проекте используется своя аннотация, а не @KafkaHandler из библиотеки Spring;

• use.standart.kafkahandler.annotation = true/false - вместе с аннотацией из property kafka.handler.annotaion будет/не будет использоваться и @KafkaHandler из библиотеки Spring (значение по умолчанию false);

• kafka.access.token.header = SERVICE_ACCESS_TOKEN - указывается наименование header, в котором присылаете токен авторизации (значение по умолчанию - SERVICE_ACCESS_TOKEN).

• language = rus - язык генерации дополнительной информации. Возможные значения: eng, rus (значение по умолчанию - eng).

Выводы

Использование принципов API First при разработке асинхронных интерфейсов решает сразу несколько проблем:

• сервисы вовремя узнают об изменении формата вашего API;

• спецификация вашего API находится в актуальном состоянии - без актуальной спецификации вы не сможете начать разработку;

• уменьшает время простоя команды - позволяет распараллелить разработку. 

Следовать этим принципам можно и без каких-либо технических средств. Но зачем, когда с их помощью можно сократить большое количество рутинной работы - создание обработчиков событий, создание моделей (Data Transfer Object). 

Что вам дает подключение AxenAPI библиотеки и использование axenapi-generator-plugin:

• вы всегда сможете отдать ссылку на актуальную спецификацию вашего API или артефакт, содержащий спецификацию вашего API;

• спецификация вашего API становится частью артефакта - вы гарантируете то, что ваше приложение реализует API, описанный в спецификации;

• появляется версионирование спецификации вашего API.

Как поучаствовать в тестировании и разработке AxenAPI

AxenAPI полностью открыт. Вы можете его найти по ссылке: https://github.com/AxenAPI

По ссылке вы найдете 4 репозитория: библиотека (axenapi-library), плагин (axenapi-gradle-plugin), генератор(axenapi-generator), примеры(axenapi-demo).

Будем благодарны за обратную связь и помощь в развитии продукта.