feat: jpa, messaging, notification, mongo, graphql 어댑터터 구현체 추가

2026-08-15 13:01:58 +09:00
parent ac874e49e6
commit 2f5d2fc219
909 changed files with 62510 additions and 6354 deletions
@@ -25,13 +25,37 @@ public class OutboundMessagePublisher implements MessagePublisher {

  @Override
  public void publish(OutboundMessage message) {
+    // The send and the observation are separate steps because they used to share a try block: a
+    // logger that threw after a successful send was caught by the same catch and reported as a
+    // publish failure. The broker had accepted the message; the only thing that failed was the
+    // record of it, and the two must not be confusable.
+    boolean sent = false;
    try {
      broker.send(message);
-      dependencyLogger.logSuccess(broker.brokerId(), DEPENDENCY_TYPE, "publish");
+      sent = true;
    } catch (Exception ex) {
      // fail-open: observe with correlationId, delegate durability to outbox/retry,
      // do NOT propagate — the core use case must still succeed.
-      dependencyLogger.logFailure(broker.brokerId(), DEPENDENCY_TYPE, "publish", ex);
+      observeQuietly(
+          () -> dependencyLogger.logFailure(broker.brokerId(), DEPENDENCY_TYPE, "publish", ex));
+    }
+    if (sent) {
+      observeQuietly(
+          () -> dependencyLogger.logSuccess(broker.brokerId(), DEPENDENCY_TYPE, "publish"));
+    }
+  }
+
+  /**
+   * Runs an observation, absorbing whatever it throws.
+   *
+   * <p>Diagnostics are non-authoritative. An appender that is out of disk must not change what the
+   * caller believes about the broker.
+   */
+  private static void observeQuietly(Runnable observation) {
+    try {
+      observation.run();
+    } catch (RuntimeException ignored) {
+      // Nothing to report it to: the reporter is what failed.
    }
  }
 }
@@ -0,0 +1,108 @@
+package dev.caskeleton.adapter.outbound.messaging.kafka;
+
+import java.util.Objects;
+
+/**
+ * One broker endpoint, parsed rather than pattern-matched.
+ *
+ * <p>The regular expression it replaces accepted several things that are not addresses. It ran
+ * against the <em>trimmed</em> value but the untrimmed original was what got stored, so {@code "
+ * kafka:9092"} passed validation and was then handed to the client with its leading space. {@code
+ * \\d{1,5}} accepts {@code 0} and {@code 99999}, neither of which is a port. And {@code [^:\\s]+}
+ * cannot express a bracketed IPv6 literal at all, so {@code [::1]:9092} — the only correct way to
+ * write an IPv6 endpoint — was rejected while {@code ::1:9092} was accepted and is ambiguous.
+ *
+ * @param host the host, without brackets for an IPv6 literal
+ * @param port the port, between 1 and 65535
+ * @param ipv6Literal whether the host was written as a bracketed IPv6 literal
+ */
+public record BrokerAddress(String host, int port, boolean ipv6Literal) {
+
+  private static final int MIN_PORT = 1;
+  private static final int MAX_PORT = 65_535;
+
+  /** Validates the parsed components. */
+  public BrokerAddress {
+    Objects.requireNonNull(host, "host must not be null");
+    if (host.isBlank()) {
+      throw new IllegalArgumentException("broker host must not be blank");
+    }
+    if (port < MIN_PORT || port > MAX_PORT) {
+      throw new IllegalArgumentException("broker port " + port + " is outside 1..65535");
+    }
+  }
+
+  /**
+   * Parses one {@code host:port} entry.
+   *
+   * @param raw the configured entry, possibly with surrounding whitespace
+   * @return the canonical address
+   * @throws IllegalArgumentException naming what is wrong with the entry
+   */
+  public static BrokerAddress parse(String raw) {
+    if (raw == null || raw.isBlank()) {
+      throw new IllegalArgumentException("a broker entry must not be blank");
+    }
+    String entry = raw.trim();
+    if (entry.startsWith("[")) {
+      int closing = entry.indexOf(']');
+      if (closing < 0 || closing + 1 >= entry.length() || entry.charAt(closing + 1) != ':') {
+        throw new IllegalArgumentException(
+            "broker entry '" + entry + "' is a bracketed host without a ':port' after the bracket");
+      }
+      String host = entry.substring(1, closing);
+      if (host.isBlank()) {
+        throw new IllegalArgumentException("broker entry '" + entry + "' has an empty host");
+      }
+      return new BrokerAddress(host, parsePort(entry, entry.substring(closing + 2)), true);
+    }
+    int separator = entry.lastIndexOf(':');
+    if (separator < 0) {
+      throw new IllegalArgumentException("broker entry '" + entry + "' is not host:port");
+    }
+    String host = entry.substring(0, separator);
+    if (host.isBlank() || host.indexOf(':') >= 0) {
+      // A bare IPv6 literal reaches here: it contains colons, and which one separates the port is
+      // not decidable. Brackets are how the ambiguity is resolved, so require them.
+      throw new IllegalArgumentException(
+          "broker entry '"
+              + entry
+              + "' is not host:port; write an IPv6 address in brackets, as [::1]:9092");
+    }
+    if (host.chars().anyMatch(Character::isWhitespace)) {
+      throw new IllegalArgumentException(
+          "broker entry '" + entry + "' has whitespace inside the host");
+    }
+    return new BrokerAddress(host, parsePort(entry, entry.substring(separator + 1)), false);
+  }
+
+  private static int parsePort(String entry, String port) {
+    if (port.isEmpty() || !port.chars().allMatch(Character::isDigit)) {
+      throw new IllegalArgumentException(
+          "broker entry '" + entry + "' does not end in a numeric port");
+    }
+    int parsed;
+    try {
+      parsed = Integer.parseInt(port);
+    } catch (NumberFormatException tooLong) {
+      throw new IllegalArgumentException(
+          "broker entry '" + entry + "' has a port outside 1..65535", tooLong);
+    }
+    if (parsed < MIN_PORT || parsed > MAX_PORT) {
+      // Reported here rather than in the constructor so the message names the offending entry: an
+      // operator reading "port 0 is invalid" against a list of nine brokers learns nothing.
+      throw new IllegalArgumentException(
+          "broker entry '" + entry + "' has port " + parsed + ", outside 1..65535");
+    }
+    return parsed;
+  }
+
+  /**
+   * The canonical {@code host:port} form, which is what the client is given.
+   *
+   * @return the canonical text
+   */
+  public String canonical() {
+    return ipv6Literal ? "[" + host + "]:" + port : host + ":" + port;
+  }
+}
@@ -1,28 +1,48 @@
 package dev.caskeleton.adapter.outbound.messaging.kafka;

 import java.util.List;
-import java.util.regex.Pattern;
 import org.springframework.boot.context.properties.ConfigurationProperties;

 /**
- * Kafka broker tuning bound from {@code app.messaging.kafka.*}. Validation is format-only ({@code
- * host:port} per entry); the "Kafka selected ⇒ brokers required" cross-field rule is enforced in
- * {@code KafkaAdapterConfig}, so an empty list is valid at bind time.
+ * Kafka broker tuning bound from {@code app.messaging.kafka.*}.
 *
- * @param brokers CSV of {@code host:port} broker endpoints (each entry format-validated)
+ * <p>Each entry is parsed into a {@link BrokerAddress} and stored in its canonical form. The
+ * previous binding validated the trimmed value with a regular expression and then stored the
+ * untrimmed original, so a configured {@code " kafka:9092"} passed the check and reached the client
+ * with its leading space; the same expression accepted port {@code 0} and port {@code 99999}, and
+ * could not express a bracketed IPv6 literal.
+ *
+ * <p>The "Kafka selected implies brokers required" cross-field rule stays in {@code
+ * KafkaAdapterConfig}, so an empty list is still valid at bind time.
+ *
+ * @param brokers CSV of {@code host:port} broker endpoints, canonicalised
 */
@ConfigurationProperties(prefix = "app.messaging.kafka")
 public record KafkaAdapterSettings(List<String> brokers) {

-  private static final Pattern HOST_PORT = Pattern.compile("^[^:\\s]+:\\d{1,5}$");
-
+  /** Parses and canonicalises every configured entry. */
  public KafkaAdapterSettings {
-    brokers = (brokers == null) ? List.of() : List.copyOf(brokers);
-    for (String broker : brokers) {
-      if (!HOST_PORT.matcher(broker.trim()).matches()) {
-        throw new IllegalArgumentException(
-            "APP_MESSAGING_KAFKA_BROKERS entry '" + broker + "' is not host:port");
-      }
-    }
+    List<String> configured = (brokers == null) ? List.of() : List.copyOf(brokers);
+    brokers =
+        configured.stream()
+            .map(
+                entry -> {
+                  try {
+                    return BrokerAddress.parse(entry).canonical();
+                  } catch (IllegalArgumentException invalid) {
+                    throw new IllegalArgumentException(
+                        "APP_MESSAGING_KAFKA_BROKERS " + invalid.getMessage(), invalid);
+                  }
+                })
+            .toList();
+  }
+
+  /**
+   * The parsed addresses, for callers that need the components rather than the text.
+   *
+   * @return one address per configured entry
+   */
+  public List<BrokerAddress> addresses() {
+    return brokers.stream().map(BrokerAddress::parse).toList();
  }
 }
@@ -37,8 +37,11 @@ public final class Slf4jOutboxRelayFailureReportAdapter implements OutboxRelayFa
      LoggingEventBuilder event =
          logger
              .atError()
-              .setCause(report.cause())
+              // No setCause: the encoder renders a Throwable's message and stack into the
+              // operational JSON, and a driver's exception text carries endpoints, statements, and
+              // occasionally credentials. The class name is a type; the code is bounded.
              .addKeyValue("error.code", report.code().code())
+              .addKeyValue("error.cause_type", report.causeType())
              .addKeyValue("error.category", report.code().category().name())
              .addKeyValue("dependency_name", dependencyName)
              .addKeyValue("dependency_type", DEPENDENCY_TYPE)
@@ -107,4 +107,40 @@ class OutboundMessagePublisherTest {
        .contains("dependency_type=\"messaging\"")
        .contains("operation=\"publish\"");
  }
+
+  @org.junit.jupiter.api.DisplayName(
+      "a logger failure after a confirmed send is not a publish failure")
+  @Test
+  void aLoggerFailureAfterAConfirmedSendIsNotAPublishFailure() {
+    // The send and the success log used to share one try block, so a logger that threw after the
+    // broker had accepted the message was caught by the failure branch and recorded as a publish
+    // failure. The broker's outcome and the record of it are different facts.
+    FakeBroker broker = new FakeBroker();
+    org.slf4j.Logger throwingLogger =
+        (org.slf4j.Logger)
+            java.lang.reflect.Proxy.newProxyInstance(
+                getClass().getClassLoader(),
+                new Class<?>[] {org.slf4j.Logger.class},
+                (proxy, method, args) -> {
+                  if (method.getName().equals("debug")) {
+                    throw new IllegalStateException("the appender is out of disk");
+                  }
+                  if (method.getReturnType() == boolean.class) {
+                    return true;
+                  }
+                  if (method.getReturnType() == String.class) {
+                    return "test.messaging";
+                  }
+                  return null;
+                });
+
+    OutboundMessagePublisher publisher =
+        new OutboundMessagePublisher(broker, new FailOpenDependencyLogger(throwingLogger));
+    OutboundMessage message = new OutboundMessage("worklog-events", "wl-1", "{}");
+
+    assertThatCode(() -> publisher.publish(message)).doesNotThrowAnyException();
+    assertThat(broker.sent)
+        .as("the message reached the broker; only the record of it failed")
+        .containsExactly(message);
+  }
 }
@@ -0,0 +1,120 @@
+package dev.caskeleton.adapter.outbound.messaging.kafka;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.util.List;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+/**
+ * What counts as a broker endpoint.
+ *
+ * <p>The regular expression that decided this ran against the trimmed value and then stored the
+ * untrimmed original, accepted port 0 and port 99999, and could not express a bracketed IPv6
+ * literal at all — so the only correct way to write an IPv6 endpoint was rejected while an
+ * ambiguous one was accepted.
+ */
+class BrokerAddressTest {
+
+  @Test
+  @DisplayName("an ordinary endpoint parses and canonicalises")
+  void anOrdinaryEndpointParses() {
+    BrokerAddress address = BrokerAddress.parse("kafka-1.internal:9092");
+
+    assertThat(address.host()).isEqualTo("kafka-1.internal");
+    assertThat(address.port()).isEqualTo(9092);
+    assertThat(address.canonical()).isEqualTo("kafka-1.internal:9092");
+  }
+
+  @Test
+  @DisplayName("surrounding whitespace is removed rather than validated away and kept")
+  void surroundingWhitespaceIsRemoved() {
+    assertThat(new KafkaAdapterSettings(List.of("  kafka:9092  ")).brokers())
+        .as("the old binding validated the trimmed value and stored the untrimmed one")
+        .containsExactly("kafka:9092");
+  }
+
+  @Test
+  @DisplayName("whitespace inside the host is refused")
+  void whitespaceInsideTheHostIsRefused() {
+    assertThatThrownBy(() -> BrokerAddress.parse("kaf ka:9092"))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("whitespace");
+  }
+
+  @Test
+  @DisplayName("the port range is 1 to 65535")
+  void thePortRangeIsOneToSixtyFiveThousand() {
+    assertThatCode(() -> BrokerAddress.parse("kafka:1")).doesNotThrowAnyException();
+    assertThatCode(() -> BrokerAddress.parse("kafka:65535")).doesNotThrowAnyException();
+
+    assertThatThrownBy(() -> BrokerAddress.parse("kafka:0"))
+        .as("port 0 asks the operating system to choose, which a client cannot connect to")
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> BrokerAddress.parse("kafka:65536"))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> BrokerAddress.parse("kafka:99999"))
+        .as("the five-digit pattern accepted this")
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  @DisplayName("a bracketed IPv6 literal is accepted and stays bracketed")
+  void aBracketedIpv6LiteralIsAccepted() {
+    BrokerAddress address = BrokerAddress.parse("[2001:db8::1]:9092");
+
+    assertThat(address.host()).isEqualTo("2001:db8::1");
+    assertThat(address.port()).isEqualTo(9092);
+    assertThat(address.ipv6Literal()).isTrue();
+    assertThat(address.canonical()).isEqualTo("[2001:db8::1]:9092");
+  }
+
+  @Test
+  @DisplayName("a bare IPv6 literal is refused because it is ambiguous")
+  void aBareIpv6LiteralIsRefused() {
+    assertThatThrownBy(() -> BrokerAddress.parse("2001:db8::1:9092"))
+        .as("which colon separates the port is not decidable, so brackets are required")
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("[::1]:9092");
+  }
+
+  @Test
+  @DisplayName("a bracketed host with no port is refused")
+  void aBracketedHostWithNoPortIsRefused() {
+    assertThatThrownBy(() -> BrokerAddress.parse("[2001:db8::1]"))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> BrokerAddress.parse("[2001:db8::1]9092"))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  @DisplayName("an entry with no port at all is refused")
+  void anEntryWithNoPortIsRefused() {
+    assertThatThrownBy(() -> BrokerAddress.parse("kafka"))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("host:port");
+    assertThatThrownBy(() -> BrokerAddress.parse("kafka:"))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> BrokerAddress.parse("kafka:http"))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> BrokerAddress.parse(":9092"))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  @DisplayName("the settings binding reports which entry is wrong")
+  void theSettingsBindingReportsWhichEntryIsWrong() {
+    assertThatThrownBy(() -> new KafkaAdapterSettings(List.of("kafka-1:9092", "kafka-2:0")))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("APP_MESSAGING_KAFKA_BROKERS")
+        .hasMessageContaining("kafka-2:0");
+  }
+
+  @Test
+  @DisplayName("an absent broker list is empty, not a failure")
+  void anAbsentBrokerListIsEmpty() {
+    assertThat(new KafkaAdapterSettings(null).brokers()).isEmpty();
+  }
+}
@@ -7,7 +7,6 @@ import static org.mockito.Mockito.when;

 import ch.qos.logback.classic.Level;
 import ch.qos.logback.classic.spi.ILoggingEvent;
-import ch.qos.logback.classic.spi.ThrowableProxy;
 import ch.qos.logback.core.read.ListAppender;
 import dev.caskeleton.application.outbox.OutboxRelayFailureReport;
 import java.time.Instant;
@@ -70,9 +69,14 @@ class Slf4jOutboxRelayFailureReportAdapterTest {
                Map.entry("aggregate_id", "agg-1"),
                Map.entry("correlation_id", "corr-1"),
                Map.entry("attempt_count", 2),
+                Map.entry("error.cause_type", "java.lang.RuntimeException"),
                Map.entry("runbook_link", "runbook://outbox/publish-failed"),
                Map.entry("next_attempt_at", "2026-07-25T01:02:03Z")));
-    assertThat(((ThrowableProxy) event.getThrowableProxy()).getThrowable()).isSameAs(cause);
+    assertThat(event.getThrowableProxy())
+        .as(
+            "the encoder renders a Throwable's message and stack into the operational JSON, and a"
+                + " driver's exception text carries endpoints, statements, and credentials")
+        .isNull();
    assertThat(event.getFormattedMessage()).doesNotContain("unsafe-exception-derived-value");
    assertThat(keyValues(event).toString())
        .doesNotContain("payload-secret", "idempotency-secret", "unsafe-exception-derived-value");
@@ -94,7 +98,7 @@ class Slf4jOutboxRelayFailureReportAdapterTest {
        .containsEntry("outcome", "DEAD")
        .containsEntry("runbook_link", "runbook://outbox/dead-letter")
        .doesNotContainKey("next_attempt_at");
-    assertThat(((ThrowableProxy) event.getThrowableProxy()).getThrowable()).isSameAs(cause);
+    assertThat(event.getThrowableProxy()).isNull();
  }

  @Test