From 2c1f18447f5880844c1891cbe22a4576029f763d Mon Sep 17 00:00:00 2001 From: ayrtons Date: Sun, 21 Jun 2026 22:50:47 -0300 Subject: [PATCH 1/2] =?UTF-8?q?Impl=20da=20resolu=C3=A7=C3=A3o?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- build.gradle | 2 +- docker-compose.yml | 48 +++++++++++++ .../labnfechallenger/config/KafkaConfig.java | 18 +++++ .../repository/NFeRepositoryMock.java | 18 +++++ .../service/KafkaConsumerPure.java | 23 +++++++ .../service/NFeProcessorService.java | 67 ++++++------------- src/main/resources/application.properties | 36 ++++++++++ 7 files changed, 165 insertions(+), 47 deletions(-) create mode 100644 docker-compose.yml create mode 100644 src/main/java/com/challengers/labnfechallenger/config/KafkaConfig.java create mode 100644 src/main/java/com/challengers/labnfechallenger/repository/NFeRepositoryMock.java create mode 100644 src/main/java/com/challengers/labnfechallenger/service/KafkaConsumerPure.java diff --git a/build.gradle b/build.gradle index 24b643a..f006363 100644 --- a/build.gradle +++ b/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') { diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..fa79f2f --- /dev/null +++ b/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" \ No newline at end of file diff --git a/src/main/java/com/challengers/labnfechallenger/config/KafkaConfig.java b/src/main/java/com/challengers/labnfechallenger/config/KafkaConfig.java new file mode 100644 index 0000000..640b797 --- /dev/null +++ b/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(); + } +} diff --git a/src/main/java/com/challengers/labnfechallenger/repository/NFeRepositoryMock.java b/src/main/java/com/challengers/labnfechallenger/repository/NFeRepositoryMock.java new file mode 100644 index 0000000..9057112 --- /dev/null +++ b/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 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"); + } +} diff --git a/src/main/java/com/challengers/labnfechallenger/service/KafkaConsumerPure.java b/src/main/java/com/challengers/labnfechallenger/service/KafkaConsumerPure.java new file mode 100644 index 0000000..2d8c9a8 --- /dev/null +++ b/src/main/java/com/challengers/labnfechallenger/service/KafkaConsumerPure.java @@ -0,0 +1,23 @@ +package com.challengers.labnfechallenger.service; + +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.stereotype.Component; + +@Component +public class KafkaConsumerPure { + + @KafkaListener(topics = "nfe-recebidas", groupId = "nfe-group", concurrency = "5") + public void escutarMensagens(ConsumerRecord record) { + String protocolo = record.key(); + System.out.println("[KAFKA WORKER] Segurando a thread para simular processamento... Protocolo: " + protocolo); + + try { + Thread.sleep(1000); + + System.out.println("[KAFKA WORKER] Sucesso ao processar: " + protocolo); + } catch (Exception e) { + System.err.println("Erro: " + e.getMessage()); + } + } +} \ No newline at end of file diff --git a/src/main/java/com/challengers/labnfechallenger/service/NFeProcessorService.java b/src/main/java/com/challengers/labnfechallenger/service/NFeProcessorService.java index 21c8e61..0df263e 100644 --- a/src/main/java/com/challengers/labnfechallenger/service/NFeProcessorService.java +++ b/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 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 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; + } } -} +} \ No newline at end of file diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index 70ab23e..dfccc43 100644 --- a/src/main/resources/application.properties +++ b/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 \ No newline at end of file From e526b010995a8251563fa7cae089f993810c7379 Mon Sep 17 00:00:00 2001 From: ayrtons Date: Wed, 24 Jun 2026 13:44:18 -0300 Subject: [PATCH 2/2] =?UTF-8?q?Impl=20da=20resolu=C3=A7=C3=A3o?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/KafkaConsumerPure.java | 32 +++++++++++++++++-- 1 file changed, 29 insertions(+), 3 deletions(-) diff --git a/src/main/java/com/challengers/labnfechallenger/service/KafkaConsumerPure.java b/src/main/java/com/challengers/labnfechallenger/service/KafkaConsumerPure.java index 2d8c9a8..a4e4346 100644 --- a/src/main/java/com/challengers/labnfechallenger/service/KafkaConsumerPure.java +++ b/src/main/java/com/challengers/labnfechallenger/service/KafkaConsumerPure.java @@ -3,6 +3,15 @@ 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 { @@ -10,14 +19,31 @@ public class KafkaConsumerPure { @KafkaListener(topics = "nfe-recebidas", groupId = "nfe-group", concurrency = "5") public void escutarMensagens(ConsumerRecord record) { String protocolo = record.key(); - System.out.println("[KAFKA WORKER] Segurando a thread para simular processamento... Protocolo: " + protocolo); + System.out.println("[KAFKA WORKER] Iniciando processamento Protocolo: " + protocolo); try { - Thread.sleep(1000); + String xmlConteudo = record.value(); + Document doc = parseXml(xmlConteudo); - System.out.println("[KAFKA WORKER] Sucesso ao processar: " + protocolo); + 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))); + } } \ No newline at end of file