chore: separate kafka consumer dlq topic

modules/sinks/sink-kafka/src/main/java/com/lingniu/ingest/sink/kafka/KafkaSinkAutoConfiguration.java

@@ -80,7 +80,7 @@ public class KafkaSinkAutoConfiguration {
     @ConditionalOnMissingBean
     public KafkaEnvelopeDeadLetterSink kafkaEnvelopeDeadLetterSink(KafkaProducer<String, byte[]> producer,
                                                                    KafkaSinkProperties props) {
-        return new KafkaEnvelopeDeadLetterSink(producer, props.getTopics().getDlq());
+        return new KafkaEnvelopeDeadLetterSink(producer, props.deadLetterTopic());
     }
 
 }

modules/sinks/sink-kafka/src/main/java/com/lingniu/ingest/sink/kafka/KafkaSinkProperties.java

@@ -51,6 +51,14 @@ public class KafkaSinkProperties {
     public Consumer getConsumer() { return consumer; }
     public void setConsumer(Consumer consumer) { this.consumer = consumer; }
 
+    public String deadLetterTopic() {
+        String consumerDlq = consumer == null ? "" : consumer.getDlqTopic();
+        if (consumerDlq != null && !consumerDlq.isBlank()) {
+            return consumerDlq;
+        }
+        return topics.getDlq();
+    }
+
     public static class Topics {
         /** 实时遥测事件 topic；GB32960 RAW-only 架构下可逐步弱化该 topic。 */
         private String realtime = "vehicle.event.gb32960.v1";
@@ -91,6 +99,8 @@ public class KafkaSinkProperties {
         private int maxPollRecords = 500;
         private int maxPollIntervalMillis = 1800000;
         private int concurrency = 1;
+        /** 消费处理失败时的死信 topic；未配置时兼容使用 producer topics.dlq。 */
+        private String dlqTopic = "";
         /**
          * Processor bean name -> Kafka binding。
          *
@@ -117,6 +127,8 @@ public class KafkaSinkProperties {
         public void setMaxPollIntervalMillis(int maxPollIntervalMillis) { this.maxPollIntervalMillis = maxPollIntervalMillis; }
         public int getConcurrency() { return concurrency; }
         public void setConcurrency(int concurrency) { this.concurrency = concurrency; }
+        public String getDlqTopic() { return dlqTopic; }
+        public void setDlqTopic(String dlqTopic) { this.dlqTopic = dlqTopic; }
         public Map<String, Binding> getBindings() { return bindings; }
         public void setBindings(Map<String, Binding> bindings) { this.bindings = bindings; }
     }

modules/sinks/sink-kafka/src/test/java/com/lingniu/ingest/sink/kafka/KafkaSinkConsumerAutoConfigurationTest.java

@@ -11,6 +11,8 @@ import org.springframework.boot.test.context.runner.ApplicationContextRunner;
 import org.springframework.boot.test.system.CapturedOutput;
 import org.springframework.boot.test.system.OutputCaptureExtension;
 
+import java.lang.reflect.Field;
+
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Mockito.mock;
 
@@ -44,8 +46,29 @@ class KafkaSinkConsumerAutoConfigurationTest {
         assertThat(output).doesNotContain("114.55.58.251");
     }
 
+    @Test
+    void consumerDlqTopicOverridesProducerTopicDefaultsForDeadLetterSink() {
+        contextRunner
+                .withPropertyValues("lingniu.ingest.sink.kafka.consumer.dlq-topic=vehicle.dlq.consumer.v1")
+                .run(context -> {
+                    assertThat(context).hasSingleBean(KafkaEnvelopeDeadLetterSink.class);
+                    assertThat(deadLetterTopic(context.getBean(KafkaEnvelopeDeadLetterSink.class)))
+                            .isEqualTo("vehicle.dlq.consumer.v1");
+                });
+    }
+
     @SuppressWarnings("unchecked")
     private static KafkaProducer<String, byte[]> kafkaProducer() {
         return mock(KafkaProducer.class);
     }
+
+    private static String deadLetterTopic(KafkaEnvelopeDeadLetterSink sink) {
+        try {
+            Field field = KafkaEnvelopeDeadLetterSink.class.getDeclaredField("topic");
+            field.setAccessible(true);
+            return (String) field.get(sink);
+        } catch (ReflectiveOperationException e) {
+            throw new AssertionError(e);
+        }
+    }
 }