Se você estiver usando o Serviço de envio de mensagens Hermes em sua instância, deverá configurar a Estrutura de notificação de eventos do produtor para o barramento de mensagens do Kafka.
Antes de Iniciar
Função necessária: administrador
O Serviço de envio de mensagens Hermes deve ser ativado. Para obter detalhes, consulte Hermes Messaging Service activation.
O snippet de código a seguir mostra os métodos
HermesEventPublisher que devem ser chamados para configurar os elementos necessários para enviar uma mensagem para um tópico usando o
Serviço de envio de mensagens Hermes. A API
do ProducerV2 é então chamada para enviar a mensagem.
publishEventToHermesKafka: function(tmfEventPayload) {
if (gs.nil(tmfEventPayload)) {
return;
}
this.publishMessageThroughProducerV2API(tmfEventPayload)
},
publishMessageThroughProducerV2API: function(message) {
if (gs.nil(message)) return;
var comptableTopicSysIds = this.fetchTopicsForMessage(message);
if (gs.nil(comptableTopicSysIds)) return;
for (var i = 0; i < comptableTopicSysIds.length; i++) {
var partitionKey = this.fetchPartionKey(message);
var headers = this.fetchMessageHeaders(message)
try {
var producer = new sn_ih_kafka.ProducerV2();
producer.send(comptableTopicSysIds[i], partitionKey, JSON.stringify(message), false, headers);
} catch (ex) {
this._logger.logErr("Publishing message to Hermes Kafka Failed, error is: " + ex);
}
}
},
Procedimento
-
Configure uma conexão segura com o Serviço de envio de mensagens Hermes.
-
Substitua a funcionalidade padrão no método HermesEventPublisherOOB - fetchTopicsForMessage (mensagem do objeto) pela lógica do seletor de tópico personalizado ao publicar uma mensagem no Serviço de envio de mensagens Hermes.
No método
HermesEventPublisher - fetchTopicsForMessage(), você deve fornecer a lógica que determina para qual tópico ou tópicos do Kafka uma mensagem especificada será enviada. Em seguida, você retorna os sys_ids desses registros de tópico em uma matriz.
Você pode selecionar um tópico com base em um atributo na carga da mensagem ou pode retornar o mesmo tópico para cada mensagem se sua implementação utilizar somente um único tópico.
O exemplo a seguir mostra como retornar o mesmo sys_id de registro de tópico para cada mensagem aprovada.
// Return the same topic for all messages if your
// implementation only supports one kind of topic
fetchTopicsForMessage: function(message) {
// Return the same sys_id for all passed messages which
// implies that all of the messages are published to the same topic.
return ['91564fc087f8351063151f473cbb35f8'];
},
O exemplo a seguir mostra como identificar tópicos para os quais uma mensagem será enviada com base no tipo de evento na mensagem.
// Fetch the topic based on the message content.
// This could be based on the message domain type or the message event type.
fetchTopicsForMessage: function(message) {
// Implement to return topic based on a message content
// Implementor may decide a topic based on 'eventType' or any other attribute.
var eventType = message.eventType;
var topicSysIdsArr = this.fetchTopicBasedOnEventType(eventType);
return topicSysIdsArr
},
-
Substitua a funcionalidade padrão no método HermesEventPublisherOOB - fetchPartionKey (mensagem do objeto) pela lógica personalizada para retornar a chave de partição a ser usada ao publicar uma mensagem no Serviço de envio de mensagens Hermes.
A lista a seguir inclui alguns tipos possíveis de chaves de partição a serem retornadas e os benefícios associados:
- Um ID exclusivo gerado automaticamente. Esse tipo de chave de partição garante que as mensagens sejam publicadas uniformemente nas partições de tópico. Esta é a implementação padrão.
- Uma chave exclusiva com base no tipo de evento de mensagem (ou em qualquer outro atributo de mensagem). Este tipo de chave de partição garante o sequenciamento de mensagens desse tipo de evento (ou outro atributo de mensagem).
- A mesma chave de partição, o que faz com que as mensagens sejam enviadas na mesma partição. Esse tipo de chave de partição garante o sequenciamento estrito de mensagens no nível do grupo de consumidores do Kafka.
O exemplo a seguir mostra como retornar um ID aleatório para cada chave de partição.
fetchPartionKey: function(message) {
// Implement it for custom partition key
// Random message keys ensure better distribution of message across partitions.
// Same message keys ensure that message always goes to the same partition.
// This ensures strict sequencing of messages at the Kafka consumer group level.
return gs.generateGUID();
},
O exemplo a seguir mostra como retornar uma chave de partição com base no tipo de evento na mensagem.
// Implementation where the partition key is decided based on the event type.
fetchPartionKey: function(message) {
var eventType = message.eventType;
var partionKey = fetchPartionKeyForMessage(eventType);
return partionKey;
},
-
Substitua a funcionalidade padrão no método HermesEventPublisherOOB - fetchMessageHeaders (mensagem do objeto) pela lógica personalizada para buscar o cabeçalho de uma mensagem especificada antes de publicar a mensagem por meio do Serviço de envio de mensagens Hermes.
Consulte a documentação
do Apache Kafka para obter mais informações sobre esses cabeçalhos.
O exemplo a seguir mostra como retornar cabeçalhos com base no tipo de evento na mensagem.
fetchMessageHeaders: function(message) {
// Returned headers are based on the event type passed in the message object.
var eventType = message.eventType;
var headerObject = fetchHeadersBasedOnEventType(eventType);
return headerObject;
},