IBM App Connect Enterprise e Event Streams

O IBM App Connect Enterprise já possui o modelo de runtime adequado para integração orientada a eventos. O laboratório ilustra como publicar um evento de negócio, como o consumir de forma segura e como manter o design de forma a ser fácil redirecionar de um broker local para o IBM Event Streams.

O laboratório é deliberadamente simples.

Iremos:

• Iniciar um broker compatível com Kafka localmente com Docker
• Criar dois tópicos Kafka
• Construir um fluxo ACE que recebe HTTP e publica um evento
• Construir um segundo fluxo ACE que consome o evento e republica uma cópia auditada
• Validar o resultado a partir da linha de comandos
• Identificar exatamente o que muda quando a mesma integração aponta para o IBM Event Streams

Para este laboratório, o runtime utiliza a imagem Docker gratuita do IBM App Connect Enterprise for Developers.

1. Objetivo do laboratório

No final deste laboratório deverá ser capaz de:

• Explicar como o ACE utiliza os nós KafkaProducer e KafkaConsumer num fluxo de eventos simples
• Separar a lógica do fluxo dos detalhes de ligação ao runtime através de uma política Kafka
• Validar o tráfego de tópicos a partir de fora do ACE
• Compreender quais as partes que permanecem inalteradas ao migrar de um laboratório local para o IBM Event Streams

2. Software necessário

Para este laboratório, necessita:

• Imagem Docker do IBM App Connect Enterprise for Developers (ibmcom/ace:latest)
• App Connect Enterprise Toolkit instalado localmente no seu computador
• Docker Engine e o plugin Docker Compose

Se o Docker não estiver instalado na sua máquina, siga os passos do Apêndice A — Instalar o Docker antes de continuar.

Se ainda não tiver o App Connect Enterprise Toolkit instalado, obtenha-o através da IBM App Connect Enterprise Developer Edition. A transferência requer uma conta IBM gratuita.

Referência IBM:

• Download ACE Developer Edition and get started

Instalar e iniciar o ACE Developer Edition no Linux

Transfira o arquivo .tar.gz a partir da ligação IBM acima e extraia-o:

tar -xzf ACE-*-LINUX64-DEVELOPER.tar.gz
cd ace-12.0.x.x

Aceite a licença e configure o ambiente:

sudo ./ace make registry global accept license silently
source /opt/ibm/ace-12.0.x.x/ace_env.sh

Inicie o Toolkit:

./ace toolkit

Na primeira execução, selecione um diretório de área de trabalho quando solicitado e, se necessário, abra a perspetiva Integration Development através de Window → Perspective → Open Perspective → Other.

3. O que iremos construir

Fluxos de mensagens e nós ACE

Um fluxo de mensagens ACE é um grafo dirigido de nós de processamento. Cada nó executa um passo: receber uma mensagem, transformá-la, encaminhá-la, enviá-la para um sistema externo. O Toolkit é a ferramenta de criação onde se colocam e ligam os nós. Após a implementação do fluxo, o runtime do ACE executa-o.

Os nós comunicam passando uma árvore de mensagens que é uma representação em memória estruturada dos cabeçalhos, corpo e ambiente da mensagem. Cada nó pode ler a árvore, modificá-la ou substituí-la antes de a passar ao nó seguinte. Os nós Compute em ESQL permitem remodelar a árvore com precisão antes de esta chegar a um nó de transporte como o KafkaProducer.

Os dois nós de transporte relevantes para este laboratório são:

• O nó KafkaProducer serializa o corpo da mensagem atual e publica-o num tópico Kafka. Como o ACE gere internamente o ciclo de vida do cliente produtor Kafka, a lógica do fluxo não gere ligações ou tentativas diretamente.
• O nó KafkaConsumer funciona como um acionador contínuo. Consulta um tópico Kafka num intervalo configurável e, para cada mensagem recebida, inicia uma nova execução do fluxo. Do ponto de vista do fluxo, é o ponto de entrada - o equivalente a um nó HTTP Input, mas orientado a eventos em vez de orientado a pedidos.

O padrão de dois fluxos e dois tópicos

O laboratório utiliza dois fluxos e dois tópicos de forma intencional. Compreender por que razão esta estrutura importa é mais relevante do que o código que a implementa.

POST /orders
  -> Compute
  -> KafkaProducer
  -> HTTPReply

KafkaConsumer(topic ace-out)
  -> Compute
  -> KafkaProducer(topic ace-audit)
  -> Trace

O Fluxo 1 liga um pedido HTTP síncrono a um evento assíncrono. O chamador recebe uma resposta imediata. O evento de negócio (a encomenda) fica agora num tópico Kafka onde qualquer número de sistemas a jusante pode consumi-lo de forma independente, ao seu próprio ritmo, sem que o chamador tenha conhecimento deles. Este é o núcleo do desacoplamento orientado a eventos: o produtor do evento e os seus consumidores não estão acoplados no tempo nem no código.

O Fluxo 2 representa um processador a jusante. Consome o evento, acrescenta os seus próprios metadados e republica uma cópia enriquecida para um segundo tópico. O segundo tópico (ace-audit) é separado do primeiro (ace-out) de modo a preservar o evento original inalterado. Qualquer sistema que precise da encomenda em bruto lê de ace-out. Qualquer sistema que precise da cópia auditada e processada lê de ace-audit. Um único tópico que transportasse ambas as formas acoplaria a lógica de auditoria a todos os consumidores que leem o original.

Isto será suficiente para ilustrar ambas as direções da interação Kafka do ACE sem tornar o artigo dependente de um ambiente extenso. A criação será feita no Toolkit e o runtime utilizado para o laboratório será o contentor a partir da imagem IBM ACE developer.

4. Criar o ambiente de laboratório local

Porquê Redpanda para um laboratório Kafka

O ficheiro compose abaixo utiliza o Redpanda como broker local em vez do Apache Kafka. O Redpanda implementa o protocolo wire do Kafka na totalidade e um cliente que fale Kafka (o que inclui os nós KafkaProducer e KafkaConsumer do ACE) não consegue distinguir a diferença ao nível do protocolo. Tópicos, grupos de consumidores, deslocamentos, semânticas de publicação e consumo, e a negociação de bootstrap que o ACE realiza ao iniciar um fluxo funcionam de forma idêntica. Cada interação que testar com o Redpanda traduz-se diretamente para o IBM Event Streams, que é ele próprio construído sobre Apache Kafka.

As razões práticas para escolher o Redpanda aqui são:

• Sem ZooKeeper. O Apache Kafka requer um conjunto ZooKeeper (ou um quórum KRaft nas versões mais recentes) a par do broker. O Redpanda é um único processo autossuficiente, o que mantém o ficheiro compose curto e o arranque rápido.
• Sem autenticação de registo. A imagem developer do Redpanda é descarregada do Docker Hub sem credenciais. A imagem developer do ACE faz o mesmo. Todo o laboratório arranca com um único docker compose up -d e sem nenhum passo de autenticação prévia.
• CLI rpk integrada. O comando rpk está incluído dentro do contentor Redpanda. Permite criar tópicos e consumir mensagens diretamente com docker compose exec sem instalar um cliente Kafka separado na sua máquina.
• Concebido para desenvolvimento local. As flags --overprovisioned e --mode dev-container no comando compose ajustam o Redpanda para um ambiente de portátil com um único nó e recursos limitados. Um broker Kafka padrão necessita de mais ajustes para se comportar bem nessas condições.

No entanto, o que o Redpanda não substitui é a configuração específica do IBM Event Streams: certificados TLS, credenciais SCRAM, tokens IAM e a integração com o registo de esquemas que uma implementação Event Streams em produção utiliza. Nada disso é relevante para o nosso laboratório. A secção 10 aborda exatamente o que muda quando se apontam os mesmos fluxos para o IBM Event Streams.

Configuração preliminar

Crie uma pasta de trabalho:

mkdir -p ~/ace-eventstreams-lab
cd ~/ace-eventstreams-lab

Crie um ficheiro docker-compose.yaml que define os nós ACE e Redpanda:

cat > docker-compose.yaml <<'EOF'
services:
  ace:
    image: ibmcom/ace:latest
    depends_on:
      - redpanda
    environment:
      LICENSE: "accept"
      ACE_SERVER_NAME: "ACESERVER"
    ports:
      - "7600:7600"
      - "7800:7800"
      - "7843:7843"

  redpanda:
    image: docker.redpanda.com/redpandadata/redpanda:latest
    command:
      - redpanda
      - start
      - --overprovisioned
      - --smp
      - "1"
      - --memory
      - "1G"
      - --reserve-memory
      - "0M"
      - --node-id
      - "0"
      - --check=false
      - --kafka-addr
      - internal://0.0.0.0:9092,external://0.0.0.0:19092
      - --advertise-kafka-addr
      - internal://redpanda:9092,external://localhost:19092
      - --pandaproxy-addr
      - internal://0.0.0.0:8082,external://0.0.0.0:18082
      - --advertise-pandaproxy-addr
      - internal://redpanda:8082,external://localhost:18082
      - --mode
      - dev-container
    ports:
      - "19092:19092"
      - "18082:18082"
EOF

Inicie o ambiente de laboratório:

docker compose up -d

Confirme que ambos os contentores estão em execução (aguarde e repita até estarem operacionais; isto demora algum tempo porque as imagens precisam de ser descarregadas):

docker compose ps

Crie os dois tópicos utilizados no laboratório e confirme que existem:

docker compose exec redpanda rpk topic create ace-out
docker compose exec redpanda rpk topic create ace-audit
docker compose exec redpanda rpk topic list

5. O projeto de política ACE e a política

O que é uma política Kafka e por que razão é importante

No ACE, uma política é um artefacto XML armazenado num Projeto de Política que externaliza a configuração de um fluxo de mensagens. Uma política Kafka contém especificamente os detalhes de ligação que os nós Kafka do ACE precisam para alcançar um broker: os endereços do servidor de bootstrap, o protocolo de segurança e, quando a segurança está ativada, credenciais, configurações TLS e mecanismos SASL.

Sem uma política, seria necessário incorporar esses detalhes diretamente em cada nó KafkaProducer e KafkaConsumer, o que tornaria a manutenção e a promoção dos fluxos entre ambientes mais difícil.

Com uma política Kafka, o nó do fluxo contém apenas a referência à política e, em runtime, o ACE resolve essa referência e utiliza os detalhes de ligação da política em vez do próprio nó. O BAR file do fluxo é o mesmo em todos os ambientes. Apenas a política implementada muda.

Um fluxo de mensagens deve codificar lógica de negócio e transformação de dados. Não deve codificar onde está o broker, que porta utiliza ou como funciona a autenticação. A política é o que mantém estas duas preocupações separadas.

Neste laboratório, a referência à política tem o seguinte aspeto:

{EventStreamsPolicyProject}:LocalKafkaPolicy

onde EventStreamsPolicyProject é o nome do Projeto de Política que cria no Toolkit e LocalKafkaPolicy é o nome da política Kafka individual dentro dele. Cada nó KafkaProducer e KafkaConsumer no laboratório utiliza a mesma referência. Se posteriormente apontar o laboratório para o IBM Event Streams, atualiza LocalKafkaPolicy ou implementa uma nova para o ambiente de destino e os fluxos não precisam de alterações.

Um detalhe de design a ter em conta: o Toolkit requer um valor no campo Bootstrap servers em cada nó antes de o guardar. Esse valor não é utilizado em runtime quando uma política está associada, pois a política substitui-o, mas deve estar presente. Para este laboratório, definimo-lo como redpanda:9092 em cada nó.

Criar o projeto e a política

No Toolkit:

1. Crie um novo Projeto de Política chamado EventStreamsPolicyProject
2. Adicione uma política Kafka chamada LocalKafkaPolicy
3. Configure-a para o broker do laboratório local:

Bootstrap servers: redpanda:9092
Security protocol: PLAINTEXT

Utilizamos redpanda:9092 e não localhost:19092 porque o runtime ACE corre dentro da rede Docker, onde alcança o Redpanda pelo nome do seu serviço. localhost:19092 continuará a ser utilizado para comunicar a partir da máquina anfitriã (para os comandos curl e rpk).

Para este laboratório local, mantemos a política intencionalmente simples. Não introduza ainda TLS, SCRAM ou IAM.

6. Criar a aplicação ACE

O que fazem os nós nestes fluxos

Antes de construir os fluxos, vale a pena ser preciso sobre o que cada tipo de nó faz.

• O nó HTTP Input abre um listener no servidor HTTP do ACE. Quando chega um pedido ao caminho configurado (/orders), o nó deserializa o corpo do pedido e coloca-o na árvore de mensagens em InputRoot.JSON, dando início à execução do fluxo. Não existe polling. A execução é acionada diretamente pela ligação HTTP de entrada.
• O nó Compute (ESQL) é o passo de transformação. ESQL é a linguagem integrada do ACE para trabalhar com a árvore de mensagens. O nó executa o módulo ESQL que lhe associar, de modo a que, após o nó Compute, a árvore de mensagens contenha o que colocou em OutputRoot. O nó seguinte no fluxo verá isso, e não o pedido original.
• O nó KafkaProducer obtém o corpo da mensagem atual em OutputRoot.JSON.Data, serializa-o e publica-o no tópico configurado. O nó gere o ciclo de vida do produtor Kafka: mantém uma ligação ao broker, trata da seleção de partição e aguarda o reconhecimento do broker antes de propagar a mensagem ao nó seguinte. O fluxo não avança para o HTTP Reply até o broker ter confirmado a publicação.
• O nó HTTP Reply envia a resposta HTTP de volta ao chamador. Quando este nó é executado, o evento produzido já está no tópico Kafka. A resposta é deliberadamente mínima neste laboratório. Num fluxo de produção incluiria tipicamente um ID de correlação ou um corpo de estado.
• O KafkaConsumer é um nó acionador. O ACE consulta o tópico configurado num intervalo configurável e, quando chega uma mensagem, o nó coloca-a na árvore de mensagens e o resto do fluxo é executado. Um fluxo KafkaConsumer é um processo contínuo em segundo plano e não tem nenhum chamador à espera de uma resposta. O ID do grupo de consumidores que configura no nó é o que o Kafka utiliza para controlar quais as mensagens que este consumidor em particular já processou. Se parar e reiniciar o fluxo, o Kafka retoma a partir do ponto onde parou para esse ID de grupo.
• Os grupos de consumidores merecem uma nota específica. Um grupo de consumidores é um grupo nomeado de um ou mais consumidores Kafka que coletivamente consomem um tópico. O Kafka atribui partições aos consumidores do grupo. Se um fluxo for escalado para múltiplas instâncias, todas as instâncias com o mesmo ID de grupo partilham o trabalho — cada mensagem é processada por exatamente uma instância. Se utilizar um ID de grupo diferente, cada instância recebe todas as mensagens de forma independente. Este laboratório utiliza um único consumidor com o ID de grupo ace-eventstreams-demo.
• O nó Trace escreve uma entrada de registo formatada no log do servidor de integração ACE. Neste laboratório regista que o fluxo de auditoria processou uma encomenda específica. É a forma mais simples de observabilidade, confirmando que o segundo fluxo foi executado e pode ser inspecionado.

Construir a aplicação

Crie uma aplicação chamada AceEventStreamsLab.

Dentro dela, construa dois fluxos de mensagens:

Fluxo 1: HTTP para Kafka

Crie um fluxo chamado HttpToKafka.msgflow com os seguintes nós:

Configuração sugerida:

• HTTP Input
  • Sufixo do caminho: /orders
• KafkaProducer
  • Nome do tópico: ace-out
  • Bootstrap servers: redpanda:9092
  • Política Kafka: {EventStreamsPolicyProject}:LocalKafkaPolicy

Na aplicação, crie um novo ficheiro ESQL chamado NormalizeOrderEvent.esql e cole este código:

CREATE COMPUTE MODULE NormalizeOrderEvent
  CREATE FUNCTION Main() RETURNS BOOLEAN
  BEGIN
    CREATE LASTCHILD OF OutputRoot DOMAIN 'JSON';

    SET OutputRoot.JSON.Data.eventType = 'OrderCreated';
    SET OutputRoot.JSON.Data.source = 'ACE';
    SET OutputRoot.JSON.Data.orderId = COALESCE(InputRoot.JSON.Data.orderId, '');
    SET OutputRoot.JSON.Data.customer = COALESCE(InputRoot.JSON.Data.customer, '');
    SET OutputRoot.JSON.Data.amount = COALESCE(InputRoot.JSON.Data.amount, 0);
    SET OutputRoot.JSON.Data.currency = COALESCE(InputRoot.JSON.Data.currency, 'EUR');
    SET OutputRoot.JSON.Data.processingNote = 'Published by ACE to topic ace-out';

    RETURN TRUE;
  END;
END MODULE;

Em seguida, abra as propriedades do nó Compute e defina o campo ESQL Module como NormalizeOrderEvent (utilize a pesquisa e seleção).

Fluxo 2: Auditoria Kafka para Kafka

Crie um segundo fluxo chamado KafkaAudit.msgflow com os seguintes nós:

Configuração sugerida:

• KafkaConsumer
  • Nome do tópico: ace-out
  • Bootstrap servers: redpanda:9092
  • ID do grupo de consumidores: ace-eventstreams-demo
  • Política Kafka: {EventStreamsPolicyProject}:LocalKafkaPolicy
• KafkaProducer
  • Nome do tópico: ace-audit
  • Bootstrap servers: redpanda:9092
  • Política Kafka: {EventStreamsPolicyProject}:LocalKafkaPolicy

Crie um segundo ficheiro ESQL chamado AddAuditMetadata.esql com o código abaixo:

CREATE COMPUTE MODULE AddAuditMetadata
  CREATE FUNCTION Main() RETURNS BOOLEAN
  BEGIN
    CREATE LASTCHILD OF OutputRoot DOMAIN 'JSON';

    SET OutputRoot.JSON.Data = InputRoot.JSON.Data;
    SET OutputRoot.JSON.Data.auditStage = 'Consumed and republished by ACE';
    SET OutputRoot.JSON.Data.auditFlow = 'KafkaAudit';

    RETURN TRUE;
  END;
END MODULE;

Em seguida, abra as propriedades do nó Compute e defina o campo ESQL Module como AddAuditMetadata.

Defina o padrão do nó Trace para algo pequeno e legível, por exemplo:

Kafka audit flow processed order ${Root.JSON.Data.orderId}

7. Implementar a aplicação

O que significa implementar no ACE

Quando implementa a partir do Toolkit, este empacota os seus fluxos e recursos num BAR file (Broker Archive) e envia-o para o servidor de integração através da API de administração. Um BAR file é um arquivo ZIP que contém os binários compilados do fluxo, quaisquer módulos ESQL e outros recursos de que o fluxo depende. O servidor de integração descomprime-o e inicia os fluxos.

Os projetos de política não são incluídos no BAR da aplicação. São implementados como artefactos separados. Isto é intencional: o mesmo BAR da aplicação pode ser executado em desenvolvimento, teste e produção — sendo a única coisa que muda entre ambientes a política implementada. Essa separação é o ponto central de externalizar os detalhes de ligação numa política.

A ordem de implementação é importante por essa razão. Quando o runtime ACE carrega um fluxo que referencia {EventStreamsPolicyProject}:LocalKafkaPolicy, procura essa política no conjunto de políticas atualmente implementadas. Se a política não estiver lá, o nó não consegue inicializar a sua ligação Kafka e o fluxo falha ao arrancar. Implementar o projeto de política primeiro garante que está presente antes de os fluxos que dependem dele serem carregados.

Antes de poder implementar a partir do Toolkit, é necessário ligá-lo ao servidor de integração que corre no contentor Docker.

No Toolkit, abra a vista Integration Servers. Clique com o botão direito e escolha Connect to an Integration Server. Introduza as seguintes propriedades de ligação:

Hostname: localhost
Port:     7600

Deixe os campos de segurança em branco para este laboratório local. A porta 7600 é a porta de administração ACE exposta pela configuração Docker Compose.

Uma vez estabelecida a ligação e visível o servidor na vista, implemente pela seguinte ordem:

1. Clique com o botão direito em EventStreamsPolicyProject e selecione Deploy. Aguarde a conclusão.
2. Clique com o botão direito em AceEventStreamsLab e selecione Deploy.

Após a implementação, o endpoint POST /orders estará disponível na porta 7800.

8. Testar o caminho de publicação

Envie um evento de negócio para o ACE:

curl -X POST http://localhost:7800/orders \
  -H "Content-Type: application/json" \
  -d '{
    "orderId": "SO-1001",
    "customer": "ACME",
    "amount": 125.50,
    "currency": "EUR"
  }'

Consuma uma mensagem diretamente do tópico de origem:

docker compose exec redpanda rpk topic consume ace-out -n 1

Deverá ver o evento normalizado produzido pelo ACE.

9. Testar o caminho de consumo e auditoria

Agora vamos validar o segundo fluxo. Consuma uma mensagem do tópico de auditoria:

docker compose exec redpanda rpk topic consume ace-audit -n 1

Se o segundo fluxo estiver ativo, deverá ver o mesmo evento de negócio com os campos adicionais:

• auditStage
• auditFlow

Isto prova que:

• O ACE publicou para Kafka
• O ACE consumiu de Kafka
• O ACE republicou um evento derivado

Trata-se já de um padrão de integração orientada a eventos válido, mesmo que a lógica de negócio seja deliberadamente pequena.

10. O que muda ao migrar para o IBM Event Streams

O ponto importante é que o design do fluxo de mensagens não precisa de ser alterado. Algumas alterações de runtime necessárias são:

1. Substituir o servidor de bootstrap local redpanda:9092 pelos servidores de bootstrap do IBM Event Streams
2. Passar de PLAINTEXT para a configuração segura adequada
3. Adicionar as credenciais necessárias e, se necessário, material de truststore
4. Reimplementar a política Kafka atualizada e as credenciais

11. Lições práticas para produção

Utilize políticas, não valores de broker fixos no código

Não coloque detalhes de broker específicos do ambiente em cada nó. Mantenha isso na política Kafka para que a transição de dev para teste e produção seja uma preocupação de runtime, não um redesenho do fluxo.

O escalonamento tem consequências na ordenação

A IBM documenta três formas principais de escalar o consumo Kafka no ACE:

• instâncias adicionais do fluxo
• múltiplos fluxos implementados no mesmo grupo de consumidores
• múltiplas ligações Kafka

Isso melhora o débito, mas pode também afetar as expetativas de ordenação. Se a ordenação for importante, trate o escalonamento e o comportamento de commit como parte do design funcional, não apenas como ajuste de runtime.

Mantenha o primeiro fluxo simples

A melhor primeira integração ACE e Event Streams não é uma complicada. É um caminho curto que prova:

• Ligação
• Escrita no tópico
• Leitura do tópico
• Transformação rastreável

Quando esse caminho estiver estável, adicione então validação de esquema, encaminhamento de erros, tentativas repetidas, tratamento de mensagens mortas e segurança mais robusta.

12. Principais conclusões

• Os nós Kafka do ACE são suficientes para construir uma primeira integração orientada a eventos limpa
• Um broker compatível com Kafka local é uma boa forma de validar a estrutura do fluxo antes de migrar para o IBM Event Streams
• A política Kafka é a articulação crítica entre a lógica do fluxo e a configuração do ambiente
• Um fluxo de publicação mais um fluxo de consumo e auditoria é um laboratório inicial melhor do que um cenário de ponta a ponta extenso
• A migração para o IBM Event Streams deve ser principalmente uma alteração de configuração, não um redesenho

Documentação IBM

Para detalhes mais aprofundados específicos da IBM, utilize estas referências oficiais:

• Using Kafka with IBM App Connect Enterprise
• Connecting IBM App Connect to IBM Event Streams
• KafkaConsumer node
• Producing messages on Kafka topics
• Consuming messages from Kafka topics
• Configuring security credentials for connecting to Kafka
• Policies overview
• Scaling messages with Kafka

Apêndice A — Instalar o Docker

Se o Docker Engine e o plugin Compose ainda não estiverem presentes na sua máquina, utilize os comandos abaixo para a sua distribuição. Execute estes passos antes de iniciar a secção 2.

Ubuntu / Debian

sudo apt-get update
sudo apt-get install -y docker.io docker-compose-plugin
sudo systemctl enable --now docker

RHEL / Rocky / Alma / CentOS Stream

sudo dnf -y install dnf-plugins-core
sudo dnf config-manager --add-repo https://download.docker.com/linux/rhel/docker-ce.repo
sudo dnf install -y docker-ce docker-ce-cli containerd.io docker-buildx-plugin docker-compose-plugin
sudo systemctl enable --now docker

Verifique a instalação:

docker version
docker compose version

Ambos os comandos devem devolver informação de versão sem erros. Se docker compose version falhar, o plugin Compose não está instalado — execute novamente o comando de instalação e verifique o resultado para identificar erros.