Transform — Go Function

Um Transform fica entre um Input e um Output e decide o que acontece com cada mensagem que passa. Hoje existe um tipo: a Go Function, onde você escreve código Go direto no console, com autocomplete e verificação de erros.

Use um Transform quando o formato que chega não é o formato que o Output espera, ou quando nem toda mensagem deve seguir adiante.

O contrato

Você edita uma função chamada process. Ela recebe uma mensagem e devolve zero, uma ou várias mensagens de saída:

func process(value string, key string, headers map[string]string, log *zap.Logger) ([]ProcessResult, error) {
	return []ProcessResult{{Value: value, Key: key, Headers: headers}}, nil
}

Parâmetro	O que é
value	O Value da mensagem que chegou
key	A Key da mensagem que chegou
headers	Os headers da mensagem que chegou
log	Para registrar mensagens que aparecem na observabilidade do node

E o que você devolve:

Retorno	Efeito
Um ProcessResult	Uma mensagem de saída
Vários ProcessResult	Uma entrada vira várias saídas
Slice vazio (nil, nil)	A mensagem é descartada e nada é publicado
Um erro	A mensagem não é confirmada e será reprocessada

Em cada ProcessResult: Value é o conteúdo, Key é a chave (string vazia = sem chave) e Headers são os headers de saída (nil repassa os headers de entrada).

Inicialização

Existe também uma função initVars, executada uma única vez quando o node sobe:

func initVars(ctx context.Context, log *zap.Logger) error {
	return nil
}

Use para o que precisa ser preparado uma vez só: compilar expressões regulares, criar um cliente HTTP, abrir um pool de conexões, carregar configuração. Devolver um erro aqui derruba o node imediatamente — de propósito: é melhor não subir do que subir mal configurado.

Exemplo completo

Criar uma pipeline monta uma integração de ponta a ponta — SQL Server CDC → Go Function → PostgreSQL — com um Transform que acrescenta um campo fixo a cada linha.

Padrões de uso

Os trechos abaixo mostram só a função process. O editor guarda o arquivo inteiro, então lembre-se de acrescentar ao bloco import o que cada exemplo usa — encoding/json, time, e assim por diante. O verificador do editor aponta o que estiver faltando.

Descartar o que não interessa

func process(value string, key string, headers map[string]string, log *zap.Logger) ([]ProcessResult, error) {
	if headers["table"] != "pedidos" {
		return nil, nil
	}
	return []ProcessResult{{Value: value, Key: key, Headers: headers}}, nil
}

Transformar o conteúdo

func process(value string, key string, headers map[string]string, log *zap.Logger) ([]ProcessResult, error) {
	var pedido map[string]any
	if err := json.Unmarshal([]byte(value), &pedido); err != nil {
		return nil, err
	}
	pedido["processado_em"] = time.Now().UTC().Format(time.RFC3339)

	saida, err := json.Marshal(pedido)
	if err != nil {
		return nil, err
	}
	return []ProcessResult{{Value: string(saida), Key: key, Headers: headers}}, nil
}

Uma entrada, várias saídas

Útil para quebrar um lote em registros individuais:

func process(value string, key string, headers map[string]string, log *zap.Logger) ([]ProcessResult, error) {
	var itens []map[string]any
	if err := json.Unmarshal([]byte(value), &itens); err != nil {
		return nil, err
	}

	saidas := make([]ProcessResult, 0, len(itens))
	for _, item := range itens {
		linha, err := json.Marshal(item)
		if err != nil {
			return nil, err
		}
		saidas = append(saidas, ProcessResult{Value: string(linha), Headers: headers})
	}
	return saidas, nil
}

Dead-letter: mensagens que você não consegue tratar

Devolver um erro faz a mensagem ser reprocessada. Se o problema é permanente — um JSON malformado, por exemplo — reprocessar cria um laço infinito: a mesma mensagem falha para sempre e trava a fila.

Para esses casos, mande a mensagem para o tópico de dead-letter ({tópico de saída}-dlq) e devolva um slice vazio:

if err := sendDlq(value, key, headers); err != nil {
	return nil, err  // não chegou ao DLQ — vale reprocessar
}
return nil, nil      // guardada no DLQ, sai do fluxo principal

A mensagem fica registrada para análise e a fila continua andando.

Headers importam

Se o Output à frente for PostgreSQL ou Snowflake, ele depende de headers específicos que a mensagem já traz — cada uma dessas páginas descreve os seus. Devolver Headers: nil repassa os que chegaram, que é o comportamento seguro. Se você montar os headers na mão, não esqueça de incluí-los.

Aruna

O editor tem a Aruna integrada para ajudar a escrever e revisar o código da função.