SOLUÇÃO DO PROBLEMA - NÃO MERGEAR

build.gradle

@@ -22,13 +22,13 @@ repositories {
 dependencies {
     implementation 'org.springframework.boot:spring-boot-starter-webmvc'
     compileOnly 'org.projectlombok:lombok'
-    developmentOnly 'org.springframework.boot:spring-boot-devtools'
     annotationProcessor 'org.projectlombok:lombok'
     providedRuntime 'org.springframework.boot:spring-boot-starter-tomcat-runtime'
     testImplementation 'org.springframework.boot:spring-boot-starter-test'
     testCompileOnly 'org.projectlombok:lombok'
     testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
     testAnnotationProcessor 'org.projectlombok:lombok'
+    implementation 'org.springframework.boot:spring-boot-starter-kafka'
 }
 
 tasks.named('test') {

docker-compose.yml

@@ -0,0 +1,48 @@
+services:
+  kafka:
+    image: confluentinc/cp-kafka:7.6.0
+    container_name: lab-kafka
+    network_mode: host
+    deploy:
+      resources:
+        limits:
+          memory: 1g
+    volumes:
+      - /etc/localtime:/etc/localtime:ro
+      - /etc/timezone:/etc/timezone:ro
+    environment:
+      TZ: "America/Sao_Paulo"
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_NODE_ID: 1
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT'
+      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://127.0.0.1:9094'
+      KAFKA_LISTENERS: 'PLAINTEXT://127.0.0.1:9094,CONTROLLER://127.0.0.1:29093'
+      KAFKA_OFFERS_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
+      KAFKA_PROCESS_ROLES: 'broker,controller'
+      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@127.0.0.1:29093'
+      KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
+      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
+      KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'
+      CLUSTER_ID: '4L62C76_QZKWg9f91mNTeA'
+      KAFKA_LOG_RETENTION_BYTES: 1073741824
+      KAFKA_LOG_SEGMENT_BYTES: 268435456
+
+  akhq:
+    image: tchiotludo/akhq
+    container_name: lab-kafka-ui
+    network_mode: host
+    volumes:
+      - /etc/localtime:/etc/localtime:ro
+      - /etc/timezone:/etc/timezone:ro
+    environment:
+      MICRONAUT_SERVER_PORT: 8089
+      AKHQ_CONFIGURATION: |
+        akhq:
+          server:
+            port: 8089
+          connections:
+            docker-kafka:
+              properties:
+                bootstrap.servers: "127.0.0.1:9094"

src/main/java/com/challengers/labnfechallenger/config/KafkaConfig.java

@@ -0,0 +1,18 @@
+package com.challengers.labnfechallenger.config;
+
+import org.apache.kafka.clients.admin.NewTopic;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.config.TopicBuilder;
+
+@Configuration
+public class KafkaConfig {
+
+    @Bean
+    public NewTopic gerarTopicoNFe() {
+        return TopicBuilder.name("nfe-recebidas")
+                .partitions(5)
+                .replicas(1)
+                .build();
+    }
+}

src/main/java/com/challengers/labnfechallenger/repository/NFeRepositoryMock.java

@@ -0,0 +1,18 @@
+package com.challengers.labnfechallenger.repository;
+
+import org.springframework.stereotype.Repository;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+@Repository
+public class NFeRepositoryMock {
+    private final ConcurrentHashMap<String, String> bancoSimulado = new ConcurrentHashMap<>();
+
+    public void salvarStatus(String protocolo, String status) {
+        bancoSimulado.put(protocolo, status);
+    }
+
+    public String buscarStatus(String protocolo) {
+        return bancoSimulado.getOrDefault(protocolo, "PROTOCOLO_NAO_ENCONTRADO");
+    }
+}

src/main/java/com/challengers/labnfechallenger/service/KafkaConsumerPure.java

@@ -0,0 +1,49 @@
+package com.challengers.labnfechallenger.service;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+import org.w3c.dom.Document;
+import org.xml.sax.InputSource;
+import org.xml.sax.SAXException;
+
+import javax.xml.parsers.DocumentBuilder;
+import javax.xml.parsers.DocumentBuilderFactory;
+import javax.xml.parsers.ParserConfigurationException;
+import java.io.IOException;
+import java.io.StringReader;
+
+@Component
+public class KafkaConsumerPure {
+
+    @KafkaListener(topics = "nfe-recebidas", groupId = "nfe-group", concurrency = "5")
+    public void escutarMensagens(ConsumerRecord<String, String> record) {
+        String protocolo = record.key();
+        System.out.println("[KAFKA WORKER] Iniciando processamento Protocolo: " + protocolo);
+
+        try {
+            String xmlConteudo = record.value();
+            Document doc = parseXml(xmlConteudo);
+
+            enviarNotaFiscalASefaz(doc, protocolo);
+        } catch (Exception e) {
+            System.err.println("Erro: " + e.getMessage());
+        }
+    }
+
+    private void enviarNotaFiscalASefaz(Document doc, String protocolo) throws InterruptedException {
+        System.out.println("[SEFAZ] Thread " + Thread.currentThread().getName() + " iniciando transmissão para o Protocolo: " + protocolo);
+
+        Thread.sleep(30000);
+
+        String versaoSefaz = doc.getXmlVersion() != null ? doc.getXmlVersion() : "4.00";
+        System.out.println("[SEFAZ] Sucesso! Doc enviado para processamento na SEFAZ. Versão do Schema: " + versaoSefaz);
+    }
+
+    private Document parseXml(String xmlConteudo) throws ParserConfigurationException, IOException, SAXException {
+        DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
+        DocumentBuilder builder = factory.newDocumentBuilder();
+
+        return builder.parse(new InputSource(new StringReader(xmlConteudo)));
+    }
+}

src/main/java/com/challengers/labnfechallenger/service/NFeProcessorService.java

@@ -1,57 +1,32 @@
 package com.challengers.labnfechallenger.service;
 
-import com.challengers.labnfechallenger.model.NFeDados;
+import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.stereotype.Service;
-import org.w3c.dom.Document;
-import org.xml.sax.InputSource;
-
-import javax.xml.parsers.DocumentBuilder;
-import javax.xml.parsers.DocumentBuilderFactory;
-import java.io.StringReader;
+import java.util.UUID;
 
 @Service
 public class NFeProcessorService {
 
-    public boolean processarEValidar(String xmlConteudo) throws Exception {
-        DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
-        DocumentBuilder builder = factory.newDocumentBuilder();
-        Document doc = builder.parse(new InputSource(new StringReader(xmlConteudo)));
-        doc.getDocumentElement().normalize();
-
-        // Extração segura pegando o primeiro elemento encontrado na árvore, independente de onde esteja
-        String cnpjEmitente = doc.getElementsByTagName("CNPJ").item(0).getTextContent();
-        String cnpjLoja     = doc.getElementsByTagName("CNPJ").item(1).getTextContent(); // O segundo CNPJ é o da Destinatária/Loja
-        String ufLoja       = doc.getElementsByTagName("UF").item(0).getTextContent();
-        String ufDestino    = doc.getElementsByTagName("UF").item(1).getTextContent();
-
-        String codProduto   = doc.getElementsByTagName("cProd").item(0).getTextContent();
-
-        // Tags exatas do padrão SEFAZ do seu arquivo
-        double vUnid        = Double.parseDouble(doc.getElementsByTagName("vUnCom").item(0).getTextContent());
-        double qCom         = Double.parseDouble(doc.getElementsByTagName("qCom").item(0).getTextContent());
-        double vProdCalc    = Double.parseDouble(doc.getElementsByTagName("vProd").item(0).getTextContent());
-        double vNF          = doc.getElementsByTagName("vNF").item(0) != null ?
-                Double.parseDouble(doc.getElementsByTagName("vNF").item(0).getTextContent()) : vProdCalc;
-
-        NFeDados nfe = new NFeDados(cnpjEmitente, cnpjLoja, ufLoja, ufDestino, codProduto, vUnid, qCom, vProdCalc, vNF);
+    private final KafkaTemplate<String, String> kafkaTemplate;
 
-        double totalProdutoCalculado = nfe.getValorUnitario() * nfe.getQuantidade();
-        boolean produtoBateComCalculo = Math.abs(totalProdutoCalculado - nfe.getValorTotalProduto()) < 0.1;
-        boolean notaBateComProduto = Math.abs(nfe.getValorTotalProduto() - nfe.getValorTotalNota()) < 0.1;
-
-        enviarNotaFiscalASefaz(doc, nfe.getCnpjEmitente());
-
-        return produtoBateComCalculo && notaBateComProduto;
+    public NFeProcessorService(KafkaTemplate<String, String> kafkaTemplate) {
+        this.kafkaTemplate = kafkaTemplate;
     }
 
-    private void enviarNotaFiscalASefaz(Document doc, String cnpjEmitente) throws InterruptedException {
-        System.out.println("[SEFAZ] Thread " + Thread.currentThread().getName() + " iniciando transmissão para o CNPJ: " + cnpjEmitente);
-
-        // Simula a lentidão da API externa da SEFAZ (30 segundos)
-        Thread.sleep(30000);
-
-        // CRÍTICO: O uso do 'doc' aqui no final força o escopo a reter a árvore DOM inteira do XML (3MB a 5MB por thread)
-        String versaoSefaz = doc.getXmlVersion() != null ? doc.getXmlVersion() : "4.00";
-        System.out.println("[SEFAZ] Sucesso! Doc enviado para processamento na SEFAZ. Versão do Schema: " + versaoSefaz);
+    public boolean processarEValidar(String xmlConteudo) {
+        if (xmlConteudo == null || xmlConteudo.trim().isEmpty()) {
+            return false;
+        }
+
+        try {
+            String protocolo = UUID.randomUUID().toString();
+
+            kafkaTemplate.send("nfe-recebidas", protocolo, xmlConteudo);
+            kafkaTemplate.flush();
+            return true;
+        } catch (Exception e) {
+            System.err.println("[PRODUCER ERROR] Falha ao publicar no Kafka: " + e.getMessage());
+            return false;
+        }
     }
-}
+}

src/main/resources/application.properties

@@ -1 +1,37 @@
 spring.application.name=LabNFeChallenger
+
+# Conexão com o Cluster Docker
+spring.kafka.bootstrap-servers=127.0.0.1:9094
+
+# Configuração do Produtor (Controller)
+spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
+spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
+
+# Configuração do Consumidor (Service)
+spring.kafka.consumer.group-id=nfe-group
+spring.kafka.consumer.auto-offset-reset=earliest
+spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
+spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
+spring.kafka.consumer.enable-auto-commit=true
+
+# Configuração do Listener
+spring.kafka.listener.auto-startup=true
+# Retorna o Tomcat para o padrão estável de threads, já que o input é instantâneo
+# 🟢 CRÍTICO: Reduz a concorrência HTTP simultânea para caber dentro de 128MB de Heap
+server.tomcat.threads.max=20
+
+server.tomcat.accept-count=200
+
+# 🟢 Força o envio IMEDIATO sem esperar agrupamento de pacotes (muda de 5ms para 0ms)
+spring.kafka.producer.linger.ms=0
+
+# 🟢 Reduz o critério de confirmação para 1 (Apenas o nó líder precisa confirmar, evita travamento)
+spring.kafka.producer.acks=1
+
+# 🟢 Limita o tempo que o produtor pode ficar travado tentando descobrir as partições
+spring.kafka.producer.properties.max.block.ms=5000
+
+# Força o envio imediato limpando o buffer de pacotes
+spring.kafka.properties.linger.ms=0
+spring.kafka.properties.acks=1
+spring.kafka.properties.max.block.ms=5000