feat: Dual 이중 버킷 유량제어 패키지 추가 (DJErp→Dual 리네임·이관)

- InflowType 열거형 추가 (ADAPTER/INTERFACE/CLIENT 타입 코드 캡슐화)
- InflowControlDAO: InflowType 기반 범용 메서드 추가, 기존 메서드 위임
- RequestProcessor: private→protected 전환, 클라이언트/어댑터/인터페이스
  유량제어 훅 메서드 5개 추가 (checkClientInflow 등)
- dual 패키지 신규 추가
  · DualCustomBucket: perSecond/threshold 이중 버킷, 차단 원인 구분
  · DualBucket: isAdapterPassDetail 등 차단 원인 반환 인터페이스
  · DualInflowControlManager: 그룹 메트릭 카운팅(TargetMetrics), 리로드 연동
  · DualRequestProcessor: 클라이언트 유량제어 + 상세 오류 메시지 훅 구현
- 단위 테스트 5종 추가 (DualCustomBucket, Manager, Metrics, RequestProcessor, Concurrency)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

src/main/java/com/eactive/eai/common/inflow/InflowControlDAO.java

@@ -41,20 +41,36 @@ public class InflowControlDAO extends BaseDAO {
 	@Autowired
 	private InflowControlGroupMappingLoader inflowControlGroupMappingLoader;
 
+	// -------------------------------------------------------------------------
+	// Enum 기반 범용 메서드 — type 코드를 직접 노출하지 않음
+	// -------------------------------------------------------------------------
+
+	public List<InflowTargetVO> getInflowList(InflowType type) throws DAOException {
+		return getInflowList(type.code());
+	}
+
+	public Map<String, InflowTargetVO> getInflowMap(InflowType type, String name) throws DAOException {
+		return getInflowMap(type.code(), name);
+	}
+
+	// -------------------------------------------------------------------------
+	// 기존 명명 메서드 — enum 기반 메서드로 위임
+	// -------------------------------------------------------------------------
+
 	public List<InflowTargetVO> getInflowAdapterList() throws DAOException {
-		return getInflowList("01");
+		return getInflowList(InflowType.ADAPTER);
 	}
 
 	public List<InflowTargetVO> getInflowInterfaceList() throws DAOException {
-		return getInflowList("02");
+		return getInflowList(InflowType.INTERFACE);
 	}
 
 	public Map<String, InflowTargetVO> getInflowTargetByAdater(String name) throws DAOException {
-		return getInflowMap("01", name);
+		return getInflowMap(InflowType.ADAPTER, name);
 	}
 
 	public Map<String, InflowTargetVO> getInflowTargetByInterface(String name) throws DAOException {
-		return getInflowMap("02", name);
+		return getInflowMap(InflowType.INTERFACE, name);
 	}
 
 	private List<InflowTargetVO> getInflowList(String type) throws DAOException {

src/main/java/com/eactive/eai/common/inflow/InflowType.java

@@ -0,0 +1,23 @@
+package com.eactive.eai.common.inflow;
+
+/**
+ * 유량제어 대상 유형.
+ * {@link InflowControlDAO}의 type 컬럼 값을 enum으로 관리하여
+ * 호출부에 코드 문자열이 노출되지 않도록 한다.
+ */
+public enum InflowType {
+
+    ADAPTER("01"),
+    INTERFACE("02"),
+    CLIENT("03");
+
+    private final String code;
+
+    InflowType(String code) {
+        this.code = code;
+    }
+
+    public String code() {
+        return code;
+    }
+}

src/main/java/com/eactive/eai/common/inflow/dual/DualBucket.java

@@ -0,0 +1,37 @@
+package com.eactive.eai.common.inflow.dual;
+
+import com.eactive.eai.common.inflow.Bucket;
+import com.eactive.eai.common.inflow.InflowTargetVO;
+
+/**
+ * Bucket 인터페이스 확장.
+ * 기존 isAdapterPass/isInterfacePass(boolean)에 더해
+ * 어느 버킷(PER_SECOND/THRESHOLD)에서 차단됐는지 반환하는 메서드 추가.
+ * 클라이언트별 유량제어 메서드도 포함.
+ */
+public interface DualBucket extends Bucket {
+
+    /**
+     * 어댑터 유량제어 체크 — 차단 원인 포함.
+     * @return null이면 통과, "PER_SECOND" 또는 "THRESHOLD"이면 해당 버킷에서 차단
+     */
+    String isAdapterPassDetail(String adapter);
+
+    /**
+     * 인터페이스 유량제어 체크 — 차단 원인 포함.
+     * @return null이면 통과, "PER_SECOND" 또는 "THRESHOLD"이면 해당 버킷에서 차단
+     */
+    String isInterfacePassDetail(String inter);
+
+    /**
+     * 클라이언트 유량제어 체크 — 차단 원인 포함.
+     * @return null이면 통과, "PER_SECOND" 또는 "THRESHOLD"이면 해당 버킷에서 차단
+     */
+    String isClientPassDetail(String clientId);
+
+    /**
+     * 클라이언트 유량제어 설정 조회 — 에러 메시지 생성용.
+     * @return 등록된 설정이 없으면 null
+     */
+    InflowTargetVO getClientInflowThreshold(String clientId);
+}

src/main/java/com/eactive/eai/common/inflow/dual/DualCustomBucket.java

@@ -0,0 +1,58 @@
+package com.eactive.eai.common.inflow.dual;
+
+import com.eactive.eai.common.inflow.InflowTargetVO;
+import io.github.bucket4j.local.LocalBucket;
+
+/**
+ * 어댑터/인터페이스 유량제어 버킷.
+ * 기존 CustomBucket(단일 버킷)과 달리 perSecond/threshold를 별도 버킷으로 분리하여
+ * 어느 한도를 초과했는지 구분 가능 (CustomGroupBucket과 동일한 이중 구조).
+ */
+public class DualCustomBucket {
+
+    public static final String RESULT_PASS               = null;
+    public static final String RESULT_BLOCKED_PER_SECOND = "PER_SECOND";
+    public static final String RESULT_BLOCKED_THRESHOLD  = "THRESHOLD";
+
+    private final LocalBucket    perSecondBucket;
+    private final LocalBucket    thresholdBucket;
+    private final InflowTargetVO inflowTargetVo;
+
+    public DualCustomBucket(LocalBucket perSecondBucket, LocalBucket thresholdBucket, InflowTargetVO inflowTargetVo) {
+        this.perSecondBucket = perSecondBucket;
+        this.thresholdBucket = thresholdBucket;
+        this.inflowTargetVo  = inflowTargetVo;
+    }
+
+    /**
+     * 두 버킷을 순서대로 소비 시도.
+     * @return null이면 통과, "PER_SECOND" 또는 "THRESHOLD"이면 해당 버킷에서 차단
+     */
+    public String tryConsume() {
+        boolean perSecondPass = (perSecondBucket == null) || perSecondBucket.tryConsume(1);
+        if (!perSecondPass) {
+            return RESULT_BLOCKED_PER_SECOND;
+        }
+
+        boolean thresholdPass = (thresholdBucket == null) || thresholdBucket.tryConsume(1);
+        if (!thresholdPass) {
+            // perSecond 토큰은 이미 소비됨 — Token Bucket 특성상 롤백 불가
+            // 초당 버킷은 빠르게 리필되므로 실질적 영향 미미
+            return RESULT_BLOCKED_THRESHOLD;
+        }
+
+        return RESULT_PASS;
+    }
+
+    public long getPerSecondAvailableTokens() {
+        return (perSecondBucket != null) ? perSecondBucket.getAvailableTokens() : -1;
+    }
+
+    public long getThresholdAvailableTokens() {
+        return (thresholdBucket != null) ? thresholdBucket.getAvailableTokens() : -1;
+    }
+
+    public LocalBucket    getPerSecondBucket() { return perSecondBucket; }
+    public LocalBucket    getThresholdBucket() { return thresholdBucket; }
+    public InflowTargetVO getInflowTargetVo()  { return inflowTargetVo; }
+}

src/main/java/com/eactive/eai/common/inflow/dual/DualInflowControlManager.java

@@ -0,0 +1,313 @@
+package com.eactive.eai.common.inflow.dual;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import com.eactive.eai.common.inflow.Bucket;
+import com.eactive.eai.common.inflow.CustomGroupBucket;
+import com.eactive.eai.common.inflow.InflowControlDAO;
+import com.eactive.eai.common.inflow.InflowControlManager;
+import com.eactive.eai.common.inflow.InflowTargetVO;
+import com.eactive.eai.common.inflow.InflowType;
+import com.eactive.eai.common.lifecycle.LifecycleException;
+import com.eactive.eai.common.message.EAIMessage;
+import com.eactive.eai.common.server.EAIServerManager;
+import com.eactive.eai.common.util.ApplicationContextProvider;
+import com.eactive.eai.common.util.Logger;
+
+import io.github.bucket4j.Bandwidth;
+import io.github.bucket4j.Bucket4j;
+import io.github.bucket4j.local.LocalBucket;
+
+/**
+ * 어댑터/인터페이스/클라이언트 버킷을 이중 구조(perSecond + threshold)로 관리하는 유량제어 관리자.
+ *
+ * <p>기존 InflowControlManager는 어댑터/인터페이스에 단일 LocalBucket(복수 Bandwidth)을 사용하여
+ * 차단 시 어느 한도(초당/기간)를 초과했는지 알 수 없었다.
+ * 이 클래스는 그룹 버킷(CustomGroupBucket)과 동일하게 두 버킷을 분리하고
+ * {@link DualBucket#isAdapterPassDetail}/{@link DualBucket#isInterfacePassDetail}로
+ * 차단 원인을 반환한다. 그룹 버킷 통과 여부는 {@link TargetMetrics}로 카운팅된다.
+ *
+ * <p>적용 방법: INFLOW.properties에 아래 한 줄 추가.
+ * <pre>inflow.control.bucket.className=com.eactive.eai.common.inflow.dual.DualInflowControlManager</pre>
+ */
+@Component
+public class DualInflowControlManager extends InflowControlManager implements DualBucket {
+
+    private static final Logger logger = Logger.getLogger(Logger.LOGGER_DEFAULT);
+
+    // InflowControlUtil이 리플렉션으로 호출하는 진입점
+    public static synchronized Bucket getBucket() {
+        return ApplicationContextProvider.getContext().getBean(DualInflowControlManager.class);
+    }
+
+    public static Boolean isTargetOfInflowControl(EAIServerManager eaiServerManager, EAIMessage eaiMessage) {
+        return InflowControlManager.isTargetOfInflowControl(eaiServerManager, eaiMessage);
+    }
+
+    @Autowired
+    private InflowControlDAO dualDao;
+
+    private volatile Map<String, DualCustomBucket> adapterBucketMap   = new ConcurrentHashMap<>();
+    private volatile Map<String, DualCustomBucket> interfaceBucketMap = new ConcurrentHashMap<>();
+    private volatile Map<String, DualCustomBucket> clientBucketMap    = new ConcurrentHashMap<>();
+
+    private final ConcurrentHashMap<String, TargetMetrics> groupMetricsMap = new ConcurrentHashMap<>();
+
+    public static class TargetMetrics {
+        public final AtomicLong allowed           = new AtomicLong();
+        public final AtomicLong rejectedPerSecond = new AtomicLong();
+        public final AtomicLong rejectedThreshold = new AtomicLong();
+        public volatile long    lastResetTime     = System.currentTimeMillis();
+
+        public void reset() {
+            allowed.set(0);
+            rejectedPerSecond.set(0);
+            rejectedThreshold.set(0);
+            lastResetTime = System.currentTimeMillis();
+        }
+    }
+
+    // -------------------------------------------------------------------------
+    // Lifecycle
+    // -------------------------------------------------------------------------
+
+    @Override
+    public void start() throws LifecycleException {
+        super.start();
+        try {
+            initAdapters();
+            initInterfaces();
+            initClients();
+        } catch (Exception e) {
+            throw new LifecycleException("DualInflowControlManager init failed: " + e.getMessage());
+        }
+    }
+
+    // -------------------------------------------------------------------------
+    // 초기화 / 리로드
+    // -------------------------------------------------------------------------
+
+    private void initAdapters() throws Exception {
+        this.adapterBucketMap = loadBucketMap(InflowType.ADAPTER);
+        if (logger.isInfo()) logger.info("DualInflowControlManager] initAdapters: " + adapterBucketMap.size() + " buckets");
+    }
+
+    private void initInterfaces() throws Exception {
+        this.interfaceBucketMap = loadBucketMap(InflowType.INTERFACE);
+        if (logger.isInfo()) logger.info("DualInflowControlManager] initInterfaces: " + interfaceBucketMap.size() + " buckets");
+    }
+
+    private void initClients() throws Exception {
+        this.clientBucketMap = loadBucketMap(InflowType.CLIENT);
+        if (logger.isInfo()) logger.info("DualInflowControlManager] initClients: " + clientBucketMap.size() + " buckets");
+    }
+
+    private Map<String, DualCustomBucket> loadBucketMap(InflowType type) throws Exception {
+        Map<String, DualCustomBucket> newMap = new HashMap<>();
+        for (InflowTargetVO vo : dualDao.getInflowList(type)) {
+            DualCustomBucket bucket = makeBucket(vo);
+            if (bucket != null) newMap.put(vo.getName(), bucket);
+        }
+        return newMap;
+    }
+
+    @Override
+    public synchronized void reloadAdapter() throws Exception {
+        super.reloadAdapter();
+        initAdapters();
+    }
+
+    @Override
+    public synchronized void reloadAdapter(String adapter) throws Exception {
+        super.reloadAdapter(adapter);
+        reloadBucketMap(adapterBucketMap, InflowType.ADAPTER, adapter);
+    }
+
+    @Override
+    public synchronized void reloadInterface() throws Exception {
+        super.reloadInterface();
+        initInterfaces();
+    }
+
+    @Override
+    public synchronized void reloadInterface(String inter) throws Exception {
+        super.reloadInterface(inter);
+        reloadBucketMap(interfaceBucketMap, InflowType.INTERFACE, inter);
+    }
+
+    public synchronized void reloadClient() throws Exception {
+        initClients();
+    }
+
+    public synchronized void reloadClient(String clientId) throws Exception {
+        reloadBucketMap(clientBucketMap, InflowType.CLIENT, clientId);
+    }
+
+    @Override
+    public synchronized void reloadGroup() throws Exception {
+        super.reloadGroup();
+        groupMetricsMap.clear();
+    }
+
+    @Override
+    public synchronized void reloadGroup(String groupId) throws Exception {
+        super.reloadGroup(groupId);
+        TargetMetrics m = groupMetricsMap.get(groupId);
+        if (m != null) m.reset();
+    }
+
+    private void reloadBucketMap(Map<String, DualCustomBucket> targetMap, InflowType type, String name) throws Exception {
+        Map<String, InflowTargetVO> updated = dualDao.getInflowMap(type, name);
+        if (updated == null) return;
+        for (InflowTargetVO vo : updated.values()) {
+            DualCustomBucket bucket = makeBucket(vo);
+            if (bucket != null) targetMap.put(vo.getName(), bucket);
+            else                targetMap.remove(vo.getName());
+        }
+    }
+
+    // -------------------------------------------------------------------------
+    // 그룹 메트릭 — isGroupPass() 카운팅 및 접근자
+    // -------------------------------------------------------------------------
+
+    @Override
+    public String isGroupPass(String groupId) {
+        String result = super.isGroupPass(groupId);
+        TargetMetrics m = groupMetricsMap.computeIfAbsent(groupId, k -> new TargetMetrics());
+        if (result == null) {
+            m.allowed.incrementAndGet();
+        } else if (CustomGroupBucket.RESULT_BLOCKED_PER_SECOND.equals(result)) {
+            m.rejectedPerSecond.incrementAndGet();
+        } else {
+            m.rejectedThreshold.incrementAndGet();
+        }
+        return result;
+    }
+
+    public TargetMetrics getGroupMetrics(String groupId) {
+        return groupMetricsMap.get(groupId);
+    }
+
+    public Map<String, TargetMetrics> getAllGroupMetrics() {
+        return Collections.unmodifiableMap(groupMetricsMap);
+    }
+
+    public void resetGroupMetrics(String groupId) {
+        TargetMetrics m = groupMetricsMap.get(groupId);
+        if (m != null) m.reset();
+    }
+
+    public void resetAllGroupMetrics() {
+        groupMetricsMap.values().forEach(TargetMetrics::reset);
+    }
+
+    // -------------------------------------------------------------------------
+    // DualBucket — 차단 원인 포함 메서드
+    // -------------------------------------------------------------------------
+
+    @Override
+    public String isAdapterPassDetail(String adapter) {
+        DualCustomBucket b = adapterBucketMap.get(adapter);
+        return (b == null) ? DualCustomBucket.RESULT_PASS : b.tryConsume();
+    }
+
+    @Override
+    public String isInterfacePassDetail(String inter) {
+        DualCustomBucket b = interfaceBucketMap.get(inter);
+        return (b == null) ? DualCustomBucket.RESULT_PASS : b.tryConsume();
+    }
+
+    @Override
+    public String isClientPassDetail(String clientId) {
+        DualCustomBucket b = clientBucketMap.get(clientId);
+        return (b == null) ? DualCustomBucket.RESULT_PASS : b.tryConsume();
+    }
+
+    @Override
+    public InflowTargetVO getClientInflowThreshold(String clientId) {
+        DualCustomBucket b = clientBucketMap.get(clientId);
+        return (b == null) ? null : b.getInflowTargetVo();
+    }
+
+    // -------------------------------------------------------------------------
+    // Bucket 인터페이스 오버라이드 — 이중 버킷으로 교체 (boolean 반환 유지)
+    // isAdapterPassDetail/isInterfacePassDetail이 토큰을 소비하므로
+    // 두 메서드를 동일 요청에서 중복 호출하면 토큰이 이중 소비됨에 주의.
+    // -------------------------------------------------------------------------
+
+    @Override
+    public boolean isAdapterPass(String adapter) {
+        return isAdapterPassDetail(adapter) == DualCustomBucket.RESULT_PASS;
+    }
+
+    @Override
+    public boolean isInterfacePass(String inter) {
+        return isInterfacePassDetail(inter) == DualCustomBucket.RESULT_PASS;
+    }
+
+    // -------------------------------------------------------------------------
+    // 모니터링용 접근자
+    // -------------------------------------------------------------------------
+
+    public DualCustomBucket getAdapterBucket(String adapter) {
+        return adapterBucketMap.get(adapter);
+    }
+
+    public DualCustomBucket getInterfaceBucket(String inter) {
+        return interfaceBucketMap.get(inter);
+    }
+
+    public Map<String, DualCustomBucket> getAdapterBucketMap() {
+        return adapterBucketMap;
+    }
+
+    public Map<String, DualCustomBucket> getInterfaceBucketMap() {
+        return interfaceBucketMap;
+    }
+
+    public DualCustomBucket getClientBucket(String clientId) {
+        return clientBucketMap.get(clientId);
+    }
+
+    public Map<String, DualCustomBucket> getClientBucketMap() {
+        return clientBucketMap;
+    }
+
+    // -------------------------------------------------------------------------
+    // 버킷 팩토리
+    // -------------------------------------------------------------------------
+
+    private DualCustomBucket makeBucket(InflowTargetVO vo) {
+        if (!InflowControlManager.isInflowTarget(vo)) {
+            return null;
+        }
+
+        LocalBucket perSecondBucket = null;
+        LocalBucket thresholdBucket = null;
+
+        if (vo.getThresholdPerSecond() > 0) {
+            perSecondBucket = Bucket4j.builder()
+                    .addLimit(Bandwidth.simple(vo.getThresholdPerSecond(), Duration.ofSeconds(1)))
+                    .build();
+        }
+
+        if (vo.getThreshold() > 0 && StringUtils.isNotBlank(vo.getThresholdTimeUnit())) {
+            thresholdBucket = Bucket4j.builder()
+                    .addLimit(Bandwidth.simple(vo.getThreshold(),
+                            InflowControlManager.getPeriod(vo.getThresholdTimeUnit())))
+                    .build();
+        }
+
+        return new DualCustomBucket(perSecondBucket, thresholdBucket, vo);
+    }
+}

src/main/java/com/eactive/eai/common/inflow/dual/DualRequestProcessor.java

@@ -0,0 +1,80 @@
+package com.eactive.eai.common.inflow.dual;
+
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.stereotype.Component;
+
+import com.eactive.eai.common.inflow.Bucket;
+import com.eactive.eai.common.inflow.InflowTargetVO;
+import com.eactive.eai.inbound.processor.RequestProcessor;
+
+/**
+ * 어댑터/인터페이스 유량제어 차단 시 어느 버킷(PER_SECOND/THRESHOLD)에서 차단됐는지
+ * 에러 메시지에 포함하는 RequestProcessor 확장.
+ *
+ * <p>DualInflowControlManager와 함께 사용:
+ * <pre>
+ * inflow.control.bucket.className=com.eactive.eai.common.inflow.dual.DualInflowControlManager
+ * </pre>
+ *
+ * <p>Spring Bean 등록 후 기존 RequestProcessor 대신 이 클래스를 어댑터에 설정한다.
+ */
+@Component("dualRequestProcessor")
+public class DualRequestProcessor extends RequestProcessor {
+
+    @Override
+    protected String checkClientInflow(Bucket bucket, String clientId) {
+        if (StringUtils.isBlank(clientId)) return null;
+        if (bucket instanceof DualBucket) {
+            return ((DualBucket) bucket).isClientPassDetail(clientId);
+        }
+        return null;
+    }
+
+    @Override
+    protected InflowTargetVO resolveClientInflowThreshold(Bucket bucket, String clientId) {
+        if (bucket instanceof DualBucket) {
+            return ((DualBucket) bucket).getClientInflowThreshold(clientId);
+        }
+        return null;
+    }
+
+    @Override
+    protected String checkAdapterInflow(Bucket bucket, String adapterGroupName) {
+        if (bucket instanceof DualBucket) {
+            return ((DualBucket) bucket).isAdapterPassDetail(adapterGroupName);
+        }
+        return super.checkAdapterInflow(bucket, adapterGroupName);
+    }
+
+    @Override
+    protected String checkInterfaceInflow(Bucket bucket, String eaiSvcCd) {
+        if (bucket instanceof DualBucket) {
+            return ((DualBucket) bucket).isInterfacePassDetail(eaiSvcCd);
+        }
+        return super.checkInterfaceInflow(bucket, eaiSvcCd);
+    }
+
+    /**
+     * 차단 원인에 따라 에러 메시지에 한도 정보를 포함.
+     * <ul>
+     *   <li>PER_SECOND: "API 호출 한도 초과 [name: Nreq/sec]"</li>
+     *   <li>THRESHOLD:  "API 호출 한도 초과 [name: Nreq/UNIT]"</li>
+     *   <li>그 외:      "API 호출 한도 초과 [name]" (기존 동작)</li>
+     * </ul>
+     */
+    @Override
+    protected String buildInflowTargetErrorMsg(InflowTargetVO inflowTargetVO, String blockedType) {
+        if (DualCustomBucket.RESULT_BLOCKED_PER_SECOND.equals(blockedType)) {
+            return String.format("API 호출 한도 초과 [%s: %dreq/sec]",
+                    inflowTargetVO.getName(),
+                    inflowTargetVO.getThresholdPerSecond());
+        }
+        if (DualCustomBucket.RESULT_BLOCKED_THRESHOLD.equals(blockedType)) {
+            return String.format("API 호출 한도 초과 [%s: %dreq/%s]",
+                    inflowTargetVO.getName(),
+                    inflowTargetVO.getThreshold(),
+                    inflowTargetVO.getThresholdTimeUnit());
+        }
+        return super.buildInflowTargetErrorMsg(inflowTargetVO, blockedType);
+    }
+}

src/main/java/com/eactive/eai/inbound/processor/RequestProcessor.java

@@ -88,8 +88,8 @@ import java.util.Map;
 
 public class RequestProcessor extends RequestProcessorSupport {
 
-	private static String instid = null;
+	protected static String instid = null;
-	private static String serverName = null;
+	protected static String serverName = null;
 	//private static boolean isPEAIServer = true;
 
 	static {
@@ -100,9 +100,9 @@ public class RequestProcessor extends RequestProcessorSupport {
 		instid = eaiServerManager.getGroupInstId();
 	}
 	
-	private static boolean httpHeaderLogMode = HttpAdapterExtraLogUtil.isHttpHeaderMode();
+	protected static boolean httpHeaderLogMode = HttpAdapterExtraLogUtil.isHttpHeaderMode();
 	
-	private ProcessVO setVO(Properties prop,long msgRcvTm) throws RequestProcessorException{
+	protected ProcessVO setVO(Properties prop,long msgRcvTm) throws RequestProcessorException{
 		String actionName        = prop.getProperty(Processor.REQUEST_ACTION);
 		String adapterGroupName  = prop.getProperty(Processor.ADAPTER_GROUP_NAME);
 		String adapterName       = prop.getProperty(Processor.ADAPTER_NAME);
@@ -1033,13 +1033,20 @@ public class RequestProcessor extends RequestProcessorSupport {
 			Bucket bucket = InflowControlUtil.getBucket();
 
 			String blockedType = "";
+			// Client 유량제어 체크 — clientId가 있는 경우에만
+			String clientBlockedType = checkClientInflow(bucket, clientId);
+			if (clientBlockedType != null) {
+				blockedType += "C";
+			}
 			// Adapter 유량제어 체크
-			if (!bucket.isAdapterPass(vo.getAdapterGroupName())) {
+			String adapterBlockedType = checkAdapterInflow(bucket, vo.getAdapterGroupName());
+			if (adapterBlockedType != null) {
 				blockedType += "A";
 			}
 			// 그룹 우선, 없으면 인터페이스 유량제어 체크
 			String groupId = bucket.getGroupIdByInterface(vo.getEaiSvcCd());
 			String groupBlockedType = null;
+			String interfaceBlockedType = null;
 			if (groupId != null) {
 				// 그룹에 속한 인터페이스 → 그룹 유량제어 체크
 				groupBlockedType = bucket.isGroupPass(groupId);
@@ -1048,7 +1055,8 @@ public class RequestProcessor extends RequestProcessorSupport {
 				}
 			} else {
 				// 그룹에 속하지 않은 인터페이스 → 인터페이스 유량제어 체크
-				if (!bucket.isInterfacePass(vo.getEaiSvcCd())) {
+				interfaceBlockedType = checkInterfaceInflow(bucket, vo.getEaiSvcCd());
+				if (interfaceBlockedType != null) {
 					blockedType += "I";
 				}
 			}
@@ -1057,7 +1065,9 @@ public class RequestProcessor extends RequestProcessorSupport {
 			if (blockedType.length() > 0) {
 				InflowTargetVO inflowTargetVO = null;
 				InflowGroupVO inflowGroupVO = null;
-				if (blockedType.startsWith("A")) {
+				if (blockedType.startsWith("C")) {
+					inflowTargetVO = resolveClientInflowThreshold(bucket, clientId);
+				} else if (blockedType.startsWith("A")) {
 					inflowTargetVO = bucket.getAdapterInflowThreashold(vo.getAdapterGroupName());
 				} else if (blockedType.contains("G")) {
 					inflowGroupVO = bucket.getGroupInflowThreshold(groupId);
@@ -1078,7 +1088,11 @@ public class RequestProcessor extends RequestProcessorSupport {
 								inflowGroupVO.getThresholdTimeUnit());
 					}
 				} else if (inflowTargetVO != null) {
-					inflowErrorMsg = String.format("API 호출 한도 초과 [%s]", inflowTargetVO.getName());
+					String targetBlockedType;
+					if (blockedType.startsWith("C"))      targetBlockedType = clientBlockedType;
+					else if (blockedType.startsWith("A")) targetBlockedType = adapterBlockedType;
+					else                                  targetBlockedType = interfaceBlockedType;
+					inflowErrorMsg = buildInflowTargetErrorMsg(inflowTargetVO, targetBlockedType);
 				} else {
 					inflowErrorMsg = "API 호출 한도 초과";
 				}
@@ -1384,7 +1398,29 @@ public class RequestProcessor extends RequestProcessorSupport {
 		}
 	}
 
-	private String getInboundErrorResponse(Properties prop, StandardMessage standardMessage, InterfaceMapper mapper
+	// 클라이언트 유량제어 체크 훅 — 기본 구현은 체크 없음, DJErpRequestProcessor가 오버라이드
+	protected String checkClientInflow(Bucket bucket, String clientId) {
+		return null;
+	}
+
+	protected InflowTargetVO resolveClientInflowThreshold(Bucket bucket, String clientId) {
+		return null;
+	}
+
+	// 어댑터/인터페이스 유량제어 체크 훅 — 서브클래스에서 오버라이드하여 차단 원인(PER_SECOND/THRESHOLD)을 반환 가능
+	protected String checkAdapterInflow(Bucket bucket, String adapterGroupName) {
+		return bucket.isAdapterPass(adapterGroupName) ? null : "BLOCKED";
+	}
+
+	protected String checkInterfaceInflow(Bucket bucket, String eaiSvcCd) {
+		return bucket.isInterfacePass(eaiSvcCd) ? null : "BLOCKED";
+	}
+
+	protected String buildInflowTargetErrorMsg(InflowTargetVO inflowTargetVO, String blockedType) {
+		return String.format("API 호출 한도 초과 [%s]", inflowTargetVO.getName());
+	}
+
+	protected String getInboundErrorResponse(Properties prop, StandardMessage standardMessage, InterfaceMapper mapper
 			, String errCode, String errMsg, String charset)  {
 		String errorResponse = null;
 
@@ -1413,7 +1449,7 @@ public class RequestProcessor extends RequestProcessorSupport {
 		return errorResponse;
 	}
 	
-    private String getInboundErrorResponseForStandard(StandardMessage standardMessage, InterfaceMapper mapper, 
+    protected String getInboundErrorResponseForStandard(StandardMessage standardMessage, InterfaceMapper mapper,
     		String errCode, String errMsg, String charset, String messageType) {
 		mapper.nextGuidSeq(standardMessage);
 		mapper.setSendRecvDivision(standardMessage, STDMessageKeys.SEND_RECV_CD_RECV);

src/test/java/com/eactive/eai/common/inflow/dual/DualCustomBucketTest.java

@@ -0,0 +1,164 @@
+package com.eactive.eai.common.inflow.dual;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.time.Duration;
+
+import org.junit.jupiter.api.Test;
+
+import com.eactive.eai.common.inflow.InflowTargetVO;
+
+import io.github.bucket4j.Bandwidth;
+import io.github.bucket4j.Bucket4j;
+import io.github.bucket4j.local.LocalBucket;
+
+class DualCustomBucketTest {
+
+    private InflowTargetVO makeVO(String name, long perSec, long threshold, String unit) {
+        InflowTargetVO vo = new InflowTargetVO();
+        vo.setName(name);
+        vo.setThresholdPerSecond(perSec);
+        vo.setThreshold(threshold);
+        vo.setThresholdTimeUnit(unit);
+        vo.setActivate(true);
+        return vo;
+    }
+
+    private LocalBucket freshBucket(long capacity) {
+        return Bucket4j.builder()
+                .addLimit(Bandwidth.simple(capacity, Duration.ofSeconds(1)))
+                .build();
+    }
+
+    private LocalBucket exhaustedBucket(long capacity) {
+        LocalBucket b = freshBucket(capacity);
+        b.tryConsume(capacity);
+        return b;
+    }
+
+    // -------------------------------------------------------------------------
+    // tryConsume — 통과 케이스
+    // -------------------------------------------------------------------------
+
+    @Test
+    void bothBucketsNull_returnsPass() {
+        DualCustomBucket bucket = new DualCustomBucket(null, null, makeVO("A", 0, 0, null));
+        assertNull(bucket.tryConsume());
+    }
+
+    @Test
+    void onlyPerSecond_withinLimit_returnsPass() {
+        DualCustomBucket bucket = new DualCustomBucket(freshBucket(5), null, makeVO("A", 5, 0, null));
+        assertNull(bucket.tryConsume());
+    }
+
+    @Test
+    void onlyThreshold_withinLimit_returnsPass() {
+        DualCustomBucket bucket = new DualCustomBucket(null, freshBucket(100), makeVO("A", 0, 100, "MIN"));
+        assertNull(bucket.tryConsume());
+    }
+
+    @Test
+    void bothBuckets_bothWithinLimit_returnsPass() {
+        DualCustomBucket bucket = new DualCustomBucket(freshBucket(10), freshBucket(100), makeVO("A", 10, 100, "MIN"));
+        assertNull(bucket.tryConsume());
+    }
+
+    // -------------------------------------------------------------------------
+    // tryConsume — 차단 케이스
+    // -------------------------------------------------------------------------
+
+    @Test
+    void onlyPerSecond_exceeded_returnsBlockedPerSecond() {
+        DualCustomBucket bucket = new DualCustomBucket(exhaustedBucket(1), null, makeVO("A", 1, 0, null));
+        assertEquals(DualCustomBucket.RESULT_BLOCKED_PER_SECOND, bucket.tryConsume());
+    }
+
+    @Test
+    void onlyThreshold_exceeded_returnsBlockedThreshold() {
+        DualCustomBucket bucket = new DualCustomBucket(null, exhaustedBucket(1), makeVO("A", 0, 1, "MIN"));
+        assertEquals(DualCustomBucket.RESULT_BLOCKED_THRESHOLD, bucket.tryConsume());
+    }
+
+    @Test
+    void perSecondExceeded_thresholdRemaining_returnsBlockedPerSecond() {
+        DualCustomBucket bucket = new DualCustomBucket(exhaustedBucket(1), freshBucket(100), makeVO("A", 1, 100, "MIN"));
+        assertEquals(DualCustomBucket.RESULT_BLOCKED_PER_SECOND, bucket.tryConsume());
+    }
+
+    @Test
+    void perSecondRemaining_thresholdExceeded_returnsBlockedThreshold() {
+        DualCustomBucket bucket = new DualCustomBucket(freshBucket(10), exhaustedBucket(1), makeVO("A", 10, 1, "MIN"));
+        assertEquals(DualCustomBucket.RESULT_BLOCKED_THRESHOLD, bucket.tryConsume());
+    }
+
+    @Test
+    void bothExceeded_returnsBlockedPerSecond() {
+        DualCustomBucket bucket = new DualCustomBucket(exhaustedBucket(1), exhaustedBucket(1), makeVO("A", 1, 1, "MIN"));
+        assertEquals(DualCustomBucket.RESULT_BLOCKED_PER_SECOND, bucket.tryConsume());
+    }
+
+    // -------------------------------------------------------------------------
+    // tryConsume — 연속 호출
+    // -------------------------------------------------------------------------
+
+    @Test
+    void consecutiveConsumes_blockedAfterCapacityExhausted() {
+        DualCustomBucket bucket = new DualCustomBucket(freshBucket(3), null, makeVO("A", 3, 0, null));
+        assertNull(bucket.tryConsume());
+        assertNull(bucket.tryConsume());
+        assertNull(bucket.tryConsume());
+        assertEquals(DualCustomBucket.RESULT_BLOCKED_PER_SECOND, bucket.tryConsume());
+    }
+
+    @Test
+    void bothBuckets_thresholdExhaustedAfterMultiplePasses() {
+        DualCustomBucket bucket = new DualCustomBucket(freshBucket(10), freshBucket(3), makeVO("A", 10, 3, "MIN"));
+        assertNull(bucket.tryConsume());
+        assertNull(bucket.tryConsume());
+        assertNull(bucket.tryConsume());
+        assertEquals(DualCustomBucket.RESULT_BLOCKED_THRESHOLD, bucket.tryConsume());
+    }
+
+    // -------------------------------------------------------------------------
+    // 가용 토큰 수 조회
+    // -------------------------------------------------------------------------
+
+    @Test
+    void nullBuckets_availableTokensReturnMinusOne() {
+        DualCustomBucket bucket = new DualCustomBucket(null, null, makeVO("A", 0, 0, null));
+        assertEquals(-1, bucket.getPerSecondAvailableTokens());
+        assertEquals(-1, bucket.getThresholdAvailableTokens());
+    }
+
+    @Test
+    void availableTokensReflectInitialCapacity() {
+        DualCustomBucket bucket = new DualCustomBucket(freshBucket(5), freshBucket(100), makeVO("A", 5, 100, "MIN"));
+        assertEquals(5, bucket.getPerSecondAvailableTokens());
+        assertEquals(100, bucket.getThresholdAvailableTokens());
+    }
+
+    @Test
+    void availableTokensDecrementAfterEachConsume() {
+        DualCustomBucket bucket = new DualCustomBucket(freshBucket(5), freshBucket(10), makeVO("A", 5, 10, "MIN"));
+        bucket.tryConsume();
+        assertEquals(4, bucket.getPerSecondAvailableTokens());
+        assertEquals(9, bucket.getThresholdAvailableTokens());
+        bucket.tryConsume();
+        assertEquals(3, bucket.getPerSecondAvailableTokens());
+        assertEquals(8, bucket.getThresholdAvailableTokens());
+    }
+
+    @Test
+    void perSecondBlocked_thresholdTokenNotConsumed() {
+        DualCustomBucket bucket = new DualCustomBucket(exhaustedBucket(1), freshBucket(10), makeVO("A", 1, 10, "MIN"));
+        long thresholdBefore = bucket.getThresholdAvailableTokens();
+        bucket.tryConsume();
+        assertEquals(thresholdBefore, bucket.getThresholdAvailableTokens());
+    }
+
+    @Test
+    void resultPassConstant_isNull() {
+        assertNull(DualCustomBucket.RESULT_PASS);
+    }
+}

src/test/java/com/eactive/eai/common/inflow/dual/DualInflowControlConcurrencyTest.java

@@ -0,0 +1,344 @@
+package com.eactive.eai.common.inflow.dual;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import com.eactive.eai.common.inflow.InflowTargetVO;
+
+import io.github.bucket4j.Bandwidth;
+import io.github.bucket4j.Bucket4j;
+import io.github.bucket4j.local.LocalBucket;
+
+/**
+ * DualCustomBucket / DualInflowControlManager 동시성 테스트.
+ *
+ * <p>모든 버킷을 Duration.ofHours(1)로 생성하여 테스트 실행 중 리필이 없도록 한다.
+ * CountDownLatch로 모든 스레드를 동시 출발시켜 race window를 최대화한다.
+ */
+class DualInflowControlConcurrencyTest {
+
+    private static final int THREAD_COUNT    = 20;
+    private static final int TRIES_PER_THREAD = 50;
+    private static final int TOTAL_TRIES     = THREAD_COUNT * TRIES_PER_THREAD; // 1000
+
+    private DualInflowControlManager manager;
+
+    @BeforeEach
+    void setUp() {
+        manager = new DualInflowControlManager();
+    }
+
+    // -------------------------------------------------------------------------
+    // Helpers
+    // -------------------------------------------------------------------------
+
+    private InflowTargetVO makeVO(String name, long perSec, long threshold, String unit) {
+        InflowTargetVO vo = new InflowTargetVO();
+        vo.setName(name);
+        vo.setThresholdPerSecond(perSec);
+        vo.setThreshold(threshold);
+        vo.setThresholdTimeUnit(unit);
+        vo.setActivate(true);
+        return vo;
+    }
+
+    private LocalBucket stableBucket(long capacity) {
+        return Bucket4j.builder()
+                .addLimit(Bandwidth.simple(capacity, Duration.ofHours(1)))
+                .build();
+    }
+
+    private boolean runConcurrently(int threads, int triesPerThread, Runnable work)
+            throws InterruptedException {
+        CountDownLatch ready = new CountDownLatch(threads);
+        CountDownLatch start = new CountDownLatch(1);
+        CountDownLatch done  = new CountDownLatch(threads);
+        ExecutorService executor = Executors.newFixedThreadPool(threads);
+
+        for (int i = 0; i < threads; i++) {
+            executor.submit(() -> {
+                ready.countDown();
+                try {
+                    start.await();
+                    for (int j = 0; j < triesPerThread; j++) {
+                        work.run();
+                    }
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                } finally {
+                    done.countDown();
+                }
+            });
+        }
+
+        ready.await();
+        start.countDown();
+        boolean completed = done.await(30, TimeUnit.SECONDS);
+        executor.shutdown();
+        return completed;
+    }
+
+    // =========================================================================
+    // DualCustomBucket 동시성 테스트
+    // =========================================================================
+
+    @Test
+    void perSecondOnly_totalPassEqualsCapacity() throws InterruptedException {
+        long capacity = 200;
+        DualCustomBucket bucket = new DualCustomBucket(
+                stableBucket(capacity), null, makeVO("A", capacity, 0, null));
+
+        AtomicInteger passCount        = new AtomicInteger();
+        AtomicInteger blockedPerSecond = new AtomicInteger();
+        AtomicInteger blockedThreshold = new AtomicInteger();
+
+        boolean done = runConcurrently(THREAD_COUNT, TRIES_PER_THREAD, () -> {
+            String result = bucket.tryConsume();
+            if (result == null)                                                   passCount.incrementAndGet();
+            else if (DualCustomBucket.RESULT_BLOCKED_PER_SECOND.equals(result))  blockedPerSecond.incrementAndGet();
+            else                                                                  blockedThreshold.incrementAndGet();
+        });
+
+        assertTrue(done, "테스트가 타임아웃되었습니다");
+        assertEquals(TOTAL_TRIES, passCount.get() + blockedPerSecond.get() + blockedThreshold.get());
+        assertEquals(capacity, passCount.get());
+        assertEquals(0, blockedThreshold.get());
+    }
+
+    @Test
+    void thresholdOnly_totalPassEqualsCapacity() throws InterruptedException {
+        long capacity = 200;
+        DualCustomBucket bucket = new DualCustomBucket(
+                null, stableBucket(capacity), makeVO("A", 0, capacity, "HOUR"));
+
+        AtomicInteger passCount        = new AtomicInteger();
+        AtomicInteger blockedThreshold = new AtomicInteger();
+
+        boolean done = runConcurrently(THREAD_COUNT, TRIES_PER_THREAD, () -> {
+            String result = bucket.tryConsume();
+            if (result == null)                                                    passCount.incrementAndGet();
+            else if (DualCustomBucket.RESULT_BLOCKED_THRESHOLD.equals(result))    blockedThreshold.incrementAndGet();
+        });
+
+        assertTrue(done, "테스트가 타임아웃되었습니다");
+        assertEquals(TOTAL_TRIES, passCount.get() + blockedThreshold.get());
+        assertEquals(capacity, passCount.get());
+    }
+
+    @Test
+    void dualBuckets_thresholdIsBottleneck_totalPassEqualsThreshold() throws InterruptedException {
+        long perSecCapacity = 2000;
+        long threshCapacity = 200;
+
+        DualCustomBucket bucket = new DualCustomBucket(
+                stableBucket(perSecCapacity), stableBucket(threshCapacity),
+                makeVO("A", perSecCapacity, threshCapacity, "HOUR"));
+
+        AtomicInteger passCount        = new AtomicInteger();
+        AtomicInteger blockedPerSecond = new AtomicInteger();
+        AtomicInteger blockedThreshold = new AtomicInteger();
+
+        boolean done = runConcurrently(THREAD_COUNT, TRIES_PER_THREAD, () -> {
+            String result = bucket.tryConsume();
+            if (result == null)                                                    passCount.incrementAndGet();
+            else if (DualCustomBucket.RESULT_BLOCKED_PER_SECOND.equals(result))   blockedPerSecond.incrementAndGet();
+            else                                                                   blockedThreshold.incrementAndGet();
+        });
+
+        assertTrue(done, "테스트가 타임아웃되었습니다");
+        assertEquals(TOTAL_TRIES, passCount.get() + blockedPerSecond.get() + blockedThreshold.get());
+        assertEquals(threshCapacity, passCount.get());
+        assertEquals(0, blockedPerSecond.get());
+    }
+
+    @Test
+    void dualBuckets_perSecondIsBottleneck_noThresholdTokenLeak() throws InterruptedException {
+        long perSecCapacity = 100;
+        long threshCapacity = 2000;
+
+        DualCustomBucket bucket = new DualCustomBucket(
+                stableBucket(perSecCapacity), stableBucket(threshCapacity),
+                makeVO("A", perSecCapacity, threshCapacity, "HOUR"));
+
+        AtomicInteger passCount        = new AtomicInteger();
+        AtomicInteger blockedPerSecond = new AtomicInteger();
+        AtomicInteger blockedThreshold = new AtomicInteger();
+
+        boolean done = runConcurrently(THREAD_COUNT, TRIES_PER_THREAD, () -> {
+            String result = bucket.tryConsume();
+            if (result == null)                                                    passCount.incrementAndGet();
+            else if (DualCustomBucket.RESULT_BLOCKED_PER_SECOND.equals(result))   blockedPerSecond.incrementAndGet();
+            else                                                                   blockedThreshold.incrementAndGet();
+        });
+
+        assertTrue(done, "테스트가 타임아웃되었습니다");
+        assertEquals(TOTAL_TRIES, passCount.get() + blockedPerSecond.get() + blockedThreshold.get());
+        assertEquals(perSecCapacity, passCount.get());
+        assertEquals(0, blockedThreshold.get());
+        assertEquals(perSecCapacity, threshCapacity - bucket.getThresholdAvailableTokens());
+    }
+
+    @Test
+    void availableTokensConsistentAfterConcurrentConsume() throws InterruptedException {
+        long capacity = 300;
+        DualCustomBucket bucket = new DualCustomBucket(
+                stableBucket(capacity), null, makeVO("A", capacity, 0, null));
+
+        AtomicInteger passCount = new AtomicInteger();
+        runConcurrently(THREAD_COUNT, TRIES_PER_THREAD, () -> {
+            if (bucket.tryConsume() == null) passCount.incrementAndGet();
+        });
+
+        long remaining = bucket.getPerSecondAvailableTokens();
+        assertEquals(capacity, passCount.get() + remaining);
+    }
+
+    // =========================================================================
+    // DualInflowControlManager 동시성 테스트
+    // =========================================================================
+
+    @Test
+    void manager_isAdapterPassDetail_concurrent_totalPassEqualsCapacity() throws InterruptedException {
+        long capacity = 200;
+        Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
+        map.put("ADAPTER_A", new DualCustomBucket(
+                stableBucket(capacity), null, makeVO("ADAPTER_A", capacity, 0, null)));
+        ReflectionTestUtils.setField(manager, "adapterBucketMap", map);
+
+        AtomicInteger passCount  = new AtomicInteger();
+        AtomicInteger blockCount = new AtomicInteger();
+
+        boolean done = runConcurrently(THREAD_COUNT, TRIES_PER_THREAD, () -> {
+            String result = manager.isAdapterPassDetail("ADAPTER_A");
+            if (result == null) passCount.incrementAndGet();
+            else                blockCount.incrementAndGet();
+        });
+
+        assertTrue(done, "테스트가 타임아웃되었습니다");
+        assertEquals(TOTAL_TRIES, passCount.get() + blockCount.get());
+        assertEquals(capacity, passCount.get());
+    }
+
+    @Test
+    void manager_isInterfacePassDetail_concurrent_totalPassEqualsCapacity() throws InterruptedException {
+        long capacity = 200;
+        Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
+        map.put("IF_001", new DualCustomBucket(
+                null, stableBucket(capacity), makeVO("IF_001", 0, capacity, "HOUR")));
+        ReflectionTestUtils.setField(manager, "interfaceBucketMap", map);
+
+        AtomicInteger passCount  = new AtomicInteger();
+        AtomicInteger blockCount = new AtomicInteger();
+
+        boolean done = runConcurrently(THREAD_COUNT, TRIES_PER_THREAD, () -> {
+            String result = manager.isInterfacePassDetail("IF_001");
+            if (result == null) passCount.incrementAndGet();
+            else                blockCount.incrementAndGet();
+        });
+
+        assertTrue(done, "테스트가 타임아웃되었습니다");
+        assertEquals(TOTAL_TRIES, passCount.get() + blockCount.get());
+        assertEquals(capacity, passCount.get());
+    }
+
+    @Test
+    void manager_volatileMapSwap_noConcurrentException() throws InterruptedException {
+        String adapterName = "ADAPTER_A";
+        int readerCount = THREAD_COUNT;
+        int swapCount   = 200;
+
+        Map<String, DualCustomBucket> initialMap = new ConcurrentHashMap<>();
+        initialMap.put(adapterName, new DualCustomBucket(
+                stableBucket(10_000), null, makeVO(adapterName, 10_000, 0, null)));
+        ReflectionTestUtils.setField(manager, "adapterBucketMap", initialMap);
+
+        AtomicReference<Throwable> error = new AtomicReference<>();
+        CountDownLatch ready = new CountDownLatch(readerCount + 1);
+        CountDownLatch start = new CountDownLatch(1);
+        CountDownLatch done  = new CountDownLatch(readerCount + 1);
+        ExecutorService executor = Executors.newFixedThreadPool(readerCount + 1);
+
+        for (int i = 0; i < readerCount; i++) {
+            executor.submit(() -> {
+                ready.countDown();
+                try {
+                    start.await();
+                    for (int j = 0; j < TRIES_PER_THREAD; j++) {
+                        manager.isAdapterPassDetail(adapterName);
+                    }
+                } catch (Throwable t) {
+                    error.compareAndSet(null, t);
+                } finally {
+                    done.countDown();
+                }
+            });
+        }
+
+        executor.submit(() -> {
+            ready.countDown();
+            try {
+                start.await();
+                for (int i = 0; i < swapCount; i++) {
+                    Map<String, DualCustomBucket> newMap = new ConcurrentHashMap<>();
+                    newMap.put(adapterName, new DualCustomBucket(
+                            stableBucket(10_000), null, makeVO(adapterName, 10_000, 0, null)));
+                    ReflectionTestUtils.setField(manager, "adapterBucketMap", newMap);
+                }
+            } catch (Throwable t) {
+                error.compareAndSet(null, t);
+            } finally {
+                done.countDown();
+            }
+        });
+
+        ready.await();
+        start.countDown();
+        boolean completed = done.await(30, TimeUnit.SECONDS);
+        executor.shutdown();
+
+        assertTrue(completed, "테스트가 타임아웃되었습니다");
+        assertNull(error.get(), error.get() != null ? "예외 발생: " + error.get() : null);
+    }
+
+    @Test
+    void manager_mixedPassAndDetailCalls_totalPassConsistent() throws InterruptedException {
+        long capacity = 300;
+        Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
+        map.put("ADAPTER_A", new DualCustomBucket(
+                stableBucket(capacity), null, makeVO("ADAPTER_A", capacity, 0, null)));
+        ReflectionTestUtils.setField(manager, "adapterBucketMap", map);
+
+        AtomicInteger passCount  = new AtomicInteger();
+        AtomicInteger blockCount = new AtomicInteger();
+        AtomicInteger callIndex  = new AtomicInteger();
+
+        boolean done = runConcurrently(THREAD_COUNT, TRIES_PER_THREAD, () -> {
+            boolean useDetail = (callIndex.getAndIncrement() % 2 == 0);
+            boolean passed;
+            if (useDetail) {
+                passed = (manager.isAdapterPassDetail("ADAPTER_A") == null);
+            } else {
+                passed = manager.isAdapterPass("ADAPTER_A");
+            }
+            if (passed) passCount.incrementAndGet();
+            else        blockCount.incrementAndGet();
+        });
+
+        assertTrue(done, "테스트가 타임아웃되었습니다");
+        assertEquals(TOTAL_TRIES, passCount.get() + blockCount.get());
+        assertEquals(capacity, passCount.get());
+    }
+}

src/test/java/com/eactive/eai/common/inflow/dual/DualInflowControlManagerMetricsTest.java

@@ -0,0 +1,318 @@
+package com.eactive.eai.common.inflow.dual;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import com.eactive.eai.common.inflow.CustomGroupBucket;
+import com.eactive.eai.common.inflow.InflowGroupVO;
+
+import io.github.bucket4j.Bandwidth;
+import io.github.bucket4j.Bucket4j;
+import io.github.bucket4j.local.LocalBucket;
+
+/**
+ * DualInflowControlManager 그룹 메트릭 단위 테스트.
+ *
+ * <p>start() 없이 groupBucketList를 ReflectionTestUtils로 직접 주입하여
+ * isGroupPass() 카운팅과 getGroupMetrics / reset 메서드를 검증한다.
+ */
+class DualInflowControlManagerMetricsTest {
+
+    private DualInflowControlManager manager;
+
+    @BeforeEach
+    void setUp() {
+        manager = new DualInflowControlManager();
+    }
+
+    // -------------------------------------------------------------------------
+    // Helpers
+    // -------------------------------------------------------------------------
+
+    private InflowGroupVO makeGroupVO(String groupId, String groupName) {
+        InflowGroupVO vo = new InflowGroupVO();
+        vo.setGroupId(groupId);
+        vo.setGroupName(groupName);
+        vo.setActivate(true);
+        return vo;
+    }
+
+    private LocalBucket stableBucket(long capacity) {
+        return Bucket4j.builder()
+                .addLimit(Bandwidth.simple(capacity, Duration.ofHours(1)))
+                .build();
+    }
+
+    private LocalBucket exhaustedBucket(long capacity) {
+        LocalBucket b = stableBucket(capacity);
+        b.tryConsume(capacity);
+        return b;
+    }
+
+    private CustomGroupBucket passBucket(String groupId) {
+        return new CustomGroupBucket(stableBucket(100), null, makeGroupVO(groupId, groupId + "-name"));
+    }
+
+    private CustomGroupBucket blockedPerSecondBucket(String groupId) {
+        return new CustomGroupBucket(exhaustedBucket(1), null, makeGroupVO(groupId, groupId + "-name"));
+    }
+
+    private CustomGroupBucket blockedThresholdBucket(String groupId) {
+        return new CustomGroupBucket(stableBucket(100), exhaustedBucket(1), makeGroupVO(groupId, groupId + "-name"));
+    }
+
+    private void injectGroupBucketList(Map<String, CustomGroupBucket> map) {
+        ReflectionTestUtils.setField(manager, "groupBucketList", map);
+    }
+
+    // =========================================================================
+    // TargetMetrics 내부 클래스
+    // =========================================================================
+
+    @Test
+    void targetMetrics_initialState_allCountersZero() {
+        DualInflowControlManager.TargetMetrics m = new DualInflowControlManager.TargetMetrics();
+
+        assertEquals(0, m.allowed.get());
+        assertEquals(0, m.rejectedPerSecond.get());
+        assertEquals(0, m.rejectedThreshold.get());
+        assertTrue(m.lastResetTime > 0);
+    }
+
+    @Test
+    void targetMetrics_reset_countersZeroAndResetTimeUpdated() throws InterruptedException {
+        DualInflowControlManager.TargetMetrics m = new DualInflowControlManager.TargetMetrics();
+        m.allowed.set(50);
+        m.rejectedPerSecond.set(10);
+        m.rejectedThreshold.set(5);
+        long before = System.currentTimeMillis();
+
+        Thread.sleep(1);
+        m.reset();
+
+        assertEquals(0, m.allowed.get());
+        assertEquals(0, m.rejectedPerSecond.get());
+        assertEquals(0, m.rejectedThreshold.get());
+        assertTrue(m.lastResetTime >= before);
+    }
+
+    // =========================================================================
+    // isGroupPass() — 카운터 증가
+    // =========================================================================
+
+    @Test
+    void isGroupPass_groupNotInBucketList_incrementsAllowedAndReturnsNull() {
+        String result = manager.isGroupPass("UNKNOWN_G");
+
+        assertNull(result);
+        DualInflowControlManager.TargetMetrics m = manager.getGroupMetrics("UNKNOWN_G");
+        assertNotNull(m);
+        assertEquals(1, m.allowed.get());
+        assertEquals(0, m.rejectedPerSecond.get());
+        assertEquals(0, m.rejectedThreshold.get());
+    }
+
+    @Test
+    void isGroupPass_pass_incrementsAllowed() {
+        Map<String, CustomGroupBucket> map = new ConcurrentHashMap<>();
+        map.put("G1", passBucket("G1"));
+        injectGroupBucketList(map);
+
+        String result = manager.isGroupPass("G1");
+
+        assertNull(result);
+        DualInflowControlManager.TargetMetrics m = manager.getGroupMetrics("G1");
+        assertEquals(1, m.allowed.get());
+        assertEquals(0, m.rejectedPerSecond.get());
+        assertEquals(0, m.rejectedThreshold.get());
+    }
+
+    @Test
+    void isGroupPass_blockedPerSecond_incrementsRejectedPerSecond() {
+        Map<String, CustomGroupBucket> map = new ConcurrentHashMap<>();
+        map.put("G1", blockedPerSecondBucket("G1"));
+        injectGroupBucketList(map);
+
+        String result = manager.isGroupPass("G1");
+
+        assertEquals(CustomGroupBucket.RESULT_BLOCKED_PER_SECOND, result);
+        DualInflowControlManager.TargetMetrics m = manager.getGroupMetrics("G1");
+        assertEquals(0, m.allowed.get());
+        assertEquals(1, m.rejectedPerSecond.get());
+        assertEquals(0, m.rejectedThreshold.get());
+    }
+
+    @Test
+    void isGroupPass_blockedThreshold_incrementsRejectedThreshold() {
+        Map<String, CustomGroupBucket> map = new ConcurrentHashMap<>();
+        map.put("G1", blockedThresholdBucket("G1"));
+        injectGroupBucketList(map);
+
+        String result = manager.isGroupPass("G1");
+
+        assertEquals(CustomGroupBucket.RESULT_BLOCKED_THRESHOLD, result);
+        DualInflowControlManager.TargetMetrics m = manager.getGroupMetrics("G1");
+        assertEquals(0, m.allowed.get());
+        assertEquals(0, m.rejectedPerSecond.get());
+        assertEquals(1, m.rejectedThreshold.get());
+    }
+
+    @Test
+    void isGroupPass_multipleCalls_countersAccumulate() {
+        Map<String, CustomGroupBucket> map = new ConcurrentHashMap<>();
+        LocalBucket b = stableBucket(5);
+        map.put("G1", new CustomGroupBucket(b, null, makeGroupVO("G1", "G1-name")));
+        injectGroupBucketList(map);
+
+        for (int i = 0; i < 8; i++) {
+            manager.isGroupPass("G1");
+        }
+
+        DualInflowControlManager.TargetMetrics m = manager.getGroupMetrics("G1");
+        assertEquals(5, m.allowed.get());
+        assertEquals(3, m.rejectedPerSecond.get());
+        assertEquals(8, m.allowed.get() + m.rejectedPerSecond.get() + m.rejectedThreshold.get());
+    }
+
+    @Test
+    void isGroupPass_differentGroups_independentMetrics() {
+        Map<String, CustomGroupBucket> map = new ConcurrentHashMap<>();
+        map.put("G1", passBucket("G1"));
+        map.put("G2", blockedPerSecondBucket("G2"));
+        injectGroupBucketList(map);
+
+        manager.isGroupPass("G1");
+        manager.isGroupPass("G1");
+        manager.isGroupPass("G2");
+
+        assertEquals(2, manager.getGroupMetrics("G1").allowed.get());
+        assertEquals(0, manager.getGroupMetrics("G1").rejectedPerSecond.get());
+        assertEquals(0, manager.getGroupMetrics("G2").allowed.get());
+        assertEquals(1, manager.getGroupMetrics("G2").rejectedPerSecond.get());
+    }
+
+    // =========================================================================
+    // getGroupMetrics / getAllGroupMetrics
+    // =========================================================================
+
+    @Test
+    void getGroupMetrics_beforeAnyCall_returnsNull() {
+        assertNull(manager.getGroupMetrics("G1"));
+    }
+
+    @Test
+    void getGroupMetrics_afterCall_returnsNonNull() {
+        manager.isGroupPass("G1");
+        assertNotNull(manager.getGroupMetrics("G1"));
+    }
+
+    @Test
+    void getAllGroupMetrics_empty_returnsEmptyMap() {
+        assertTrue(manager.getAllGroupMetrics().isEmpty());
+    }
+
+    @Test
+    void getAllGroupMetrics_afterMultipleGroups_containsAll() {
+        manager.isGroupPass("G1");
+        manager.isGroupPass("G2");
+        manager.isGroupPass("G3");
+
+        Map<String, DualInflowControlManager.TargetMetrics> all = manager.getAllGroupMetrics();
+        assertEquals(3, all.size());
+        assertTrue(all.containsKey("G1"));
+        assertTrue(all.containsKey("G2"));
+        assertTrue(all.containsKey("G3"));
+    }
+
+    @Test
+    void getAllGroupMetrics_returnsUnmodifiableView() {
+        manager.isGroupPass("G1");
+
+        Map<String, DualInflowControlManager.TargetMetrics> all = manager.getAllGroupMetrics();
+        assertThrows(UnsupportedOperationException.class, () -> all.put("NEW", null));
+    }
+
+    // =========================================================================
+    // resetGroupMetrics / resetAllGroupMetrics
+    // =========================================================================
+
+    @Test
+    void resetGroupMetrics_resetsTargetGroup() {
+        Map<String, CustomGroupBucket> map = new ConcurrentHashMap<>();
+        map.put("G1", passBucket("G1"));
+        injectGroupBucketList(map);
+
+        manager.isGroupPass("G1");
+        manager.isGroupPass("G1");
+        assertEquals(2, manager.getGroupMetrics("G1").allowed.get());
+
+        manager.resetGroupMetrics("G1");
+
+        DualInflowControlManager.TargetMetrics m = manager.getGroupMetrics("G1");
+        assertEquals(0, m.allowed.get());
+        assertEquals(0, m.rejectedPerSecond.get());
+        assertEquals(0, m.rejectedThreshold.get());
+    }
+
+    @Test
+    void resetGroupMetrics_updatesLastResetTime() throws InterruptedException {
+        manager.isGroupPass("G1");
+        long timeBefore = manager.getGroupMetrics("G1").lastResetTime;
+
+        Thread.sleep(1);
+        manager.resetGroupMetrics("G1");
+
+        assertTrue(manager.getGroupMetrics("G1").lastResetTime >= timeBefore);
+    }
+
+    @Test
+    void resetGroupMetrics_nonExistentGroup_noException() {
+        assertDoesNotThrow(() -> manager.resetGroupMetrics("NON_EXISTENT"));
+    }
+
+    @Test
+    void resetGroupMetrics_doesNotAffectOtherGroups() {
+        manager.isGroupPass("G1");
+        manager.isGroupPass("G2");
+
+        manager.resetGroupMetrics("G1");
+
+        assertEquals(0, manager.getGroupMetrics("G1").allowed.get());
+        assertEquals(1, manager.getGroupMetrics("G2").allowed.get());
+    }
+
+    @Test
+    void resetAllGroupMetrics_resetsAllGroups() {
+        manager.isGroupPass("G1");
+        manager.isGroupPass("G1");
+        manager.isGroupPass("G2");
+
+        manager.resetAllGroupMetrics();
+
+        assertEquals(0, manager.getGroupMetrics("G1").allowed.get());
+        assertEquals(0, manager.getGroupMetrics("G2").allowed.get());
+    }
+
+    @Test
+    void resetAllGroupMetrics_emptyMap_noException() {
+        assertDoesNotThrow(() -> manager.resetAllGroupMetrics());
+    }
+
+    @Test
+    void resetAllGroupMetrics_preservesKeys() {
+        manager.isGroupPass("G1");
+        manager.isGroupPass("G2");
+
+        manager.resetAllGroupMetrics();
+
+        assertNotNull(manager.getGroupMetrics("G1"));
+        assertNotNull(manager.getGroupMetrics("G2"));
+    }
+}

src/test/java/com/eactive/eai/common/inflow/dual/DualInflowControlManagerTest.java

@@ -0,0 +1,244 @@
+package com.eactive.eai.common.inflow.dual;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import com.eactive.eai.common.inflow.InflowTargetVO;
+
+import io.github.bucket4j.Bandwidth;
+import io.github.bucket4j.Bucket4j;
+import io.github.bucket4j.local.LocalBucket;
+
+/**
+ * DualInflowControlManager 단위 테스트.
+ *
+ * <p>start() 호출 없이 adapterBucketMap / interfaceBucketMap을
+ * ReflectionTestUtils로 직접 주입하여 detail 메서드만 검증한다.
+ */
+class DualInflowControlManagerTest {
+
+    private DualInflowControlManager manager;
+
+    @BeforeEach
+    void setUp() {
+        manager = new DualInflowControlManager();
+    }
+
+    // -------------------------------------------------------------------------
+    // Helpers
+    // -------------------------------------------------------------------------
+
+    private InflowTargetVO makeVO(String name) {
+        InflowTargetVO vo = new InflowTargetVO();
+        vo.setName(name);
+        vo.setThresholdPerSecond(10);
+        vo.setThreshold(100);
+        vo.setThresholdTimeUnit("MIN");
+        vo.setActivate(true);
+        return vo;
+    }
+
+    private LocalBucket freshBucket(long capacity) {
+        return Bucket4j.builder()
+                .addLimit(Bandwidth.simple(capacity, Duration.ofSeconds(1)))
+                .build();
+    }
+
+    private LocalBucket exhaustedBucket(long capacity) {
+        LocalBucket b = freshBucket(capacity);
+        b.tryConsume(capacity);
+        return b;
+    }
+
+    private DualCustomBucket passBucket(String name) {
+        return new DualCustomBucket(freshBucket(10), freshBucket(100), makeVO(name));
+    }
+
+    private DualCustomBucket blockedPerSecondBucket(String name) {
+        return new DualCustomBucket(exhaustedBucket(1), freshBucket(100), makeVO(name));
+    }
+
+    private DualCustomBucket blockedThresholdBucket(String name) {
+        return new DualCustomBucket(freshBucket(10), exhaustedBucket(1), makeVO(name));
+    }
+
+    private void injectAdapterMap(Map<String, DualCustomBucket> map) {
+        ReflectionTestUtils.setField(manager, "adapterBucketMap", map);
+    }
+
+    private void injectInterfaceMap(Map<String, DualCustomBucket> map) {
+        ReflectionTestUtils.setField(manager, "interfaceBucketMap", map);
+    }
+
+    // -------------------------------------------------------------------------
+    // isAdapterPassDetail
+    // -------------------------------------------------------------------------
+
+    @Test
+    void isAdapterPassDetail_adapterNotInMap_returnsNull() {
+        assertNull(manager.isAdapterPassDetail("UNKNOWN"));
+    }
+
+    @Test
+    void isAdapterPassDetail_adapterInMap_pass_returnsNull() {
+        Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
+        map.put("ADAPTER_A", passBucket("ADAPTER_A"));
+        injectAdapterMap(map);
+
+        assertNull(manager.isAdapterPassDetail("ADAPTER_A"));
+    }
+
+    @Test
+    void isAdapterPassDetail_adapterInMap_blockedPerSecond() {
+        Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
+        map.put("ADAPTER_A", blockedPerSecondBucket("ADAPTER_A"));
+        injectAdapterMap(map);
+
+        assertEquals(DualCustomBucket.RESULT_BLOCKED_PER_SECOND,
+                manager.isAdapterPassDetail("ADAPTER_A"));
+    }
+
+    @Test
+    void isAdapterPassDetail_adapterInMap_blockedThreshold() {
+        Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
+        map.put("ADAPTER_A", blockedThresholdBucket("ADAPTER_A"));
+        injectAdapterMap(map);
+
+        assertEquals(DualCustomBucket.RESULT_BLOCKED_THRESHOLD,
+                manager.isAdapterPassDetail("ADAPTER_A"));
+    }
+
+    // -------------------------------------------------------------------------
+    // isInterfacePassDetail
+    // -------------------------------------------------------------------------
+
+    @Test
+    void isInterfacePassDetail_interfaceNotInMap_returnsNull() {
+        assertNull(manager.isInterfacePassDetail("UNKNOWN"));
+    }
+
+    @Test
+    void isInterfacePassDetail_interfaceInMap_pass_returnsNull() {
+        Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
+        map.put("IF_001", passBucket("IF_001"));
+        injectInterfaceMap(map);
+
+        assertNull(manager.isInterfacePassDetail("IF_001"));
+    }
+
+    @Test
+    void isInterfacePassDetail_interfaceInMap_blockedPerSecond() {
+        Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
+        map.put("IF_001", blockedPerSecondBucket("IF_001"));
+        injectInterfaceMap(map);
+
+        assertEquals(DualCustomBucket.RESULT_BLOCKED_PER_SECOND,
+                manager.isInterfacePassDetail("IF_001"));
+    }
+
+    @Test
+    void isInterfacePassDetail_interfaceInMap_blockedThreshold() {
+        Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
+        map.put("IF_001", blockedThresholdBucket("IF_001"));
+        injectInterfaceMap(map);
+
+        assertEquals(DualCustomBucket.RESULT_BLOCKED_THRESHOLD,
+                manager.isInterfacePassDetail("IF_001"));
+    }
+
+    // -------------------------------------------------------------------------
+    // isAdapterPass — boolean 래퍼
+    // -------------------------------------------------------------------------
+
+    @Test
+    void isAdapterPass_notInMap_returnsTrue() {
+        assertTrue(manager.isAdapterPass("UNKNOWN"));
+    }
+
+    @Test
+    void isAdapterPass_pass_returnsTrue() {
+        Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
+        map.put("ADAPTER_A", passBucket("ADAPTER_A"));
+        injectAdapterMap(map);
+
+        assertTrue(manager.isAdapterPass("ADAPTER_A"));
+    }
+
+    @Test
+    void isAdapterPass_blocked_returnsFalse() {
+        Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
+        map.put("ADAPTER_A", blockedPerSecondBucket("ADAPTER_A"));
+        injectAdapterMap(map);
+
+        assertFalse(manager.isAdapterPass("ADAPTER_A"));
+    }
+
+    // -------------------------------------------------------------------------
+    // isInterfacePass — boolean 래퍼
+    // -------------------------------------------------------------------------
+
+    @Test
+    void isInterfacePass_notInMap_returnsTrue() {
+        assertTrue(manager.isInterfacePass("UNKNOWN"));
+    }
+
+    @Test
+    void isInterfacePass_pass_returnsTrue() {
+        Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
+        map.put("IF_001", passBucket("IF_001"));
+        injectInterfaceMap(map);
+
+        assertTrue(manager.isInterfacePass("IF_001"));
+    }
+
+    @Test
+    void isInterfacePass_blocked_returnsFalse() {
+        Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
+        map.put("IF_001", blockedThresholdBucket("IF_001"));
+        injectInterfaceMap(map);
+
+        assertFalse(manager.isInterfacePass("IF_001"));
+    }
+
+    // -------------------------------------------------------------------------
+    // 모니터링 접근자
+    // -------------------------------------------------------------------------
+
+    @Test
+    void getAdapterBucket_returnsCorrectBucket() {
+        Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
+        DualCustomBucket expected = passBucket("ADAPTER_A");
+        map.put("ADAPTER_A", expected);
+        injectAdapterMap(map);
+
+        assertSame(expected, manager.getAdapterBucket("ADAPTER_A"));
+        assertNull(manager.getAdapterBucket("UNKNOWN"));
+    }
+
+    @Test
+    void getInterfaceBucket_returnsCorrectBucket() {
+        Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
+        DualCustomBucket expected = passBucket("IF_001");
+        map.put("IF_001", expected);
+        injectInterfaceMap(map);
+
+        assertSame(expected, manager.getInterfaceBucket("IF_001"));
+        assertNull(manager.getInterfaceBucket("UNKNOWN"));
+    }
+
+    @Test
+    void getAdapterBucketMap_containsInjectedEntries() {
+        Map<String, DualCustomBucket> injected = new ConcurrentHashMap<>();
+        injected.put("ADAPTER_A", passBucket("ADAPTER_A"));
+        injectAdapterMap(injected);
+
+        assertTrue(manager.getAdapterBucketMap().containsKey("ADAPTER_A"));
+    }
+}

src/test/java/com/eactive/eai/common/inflow/dual/DualRequestProcessorTest.java

@@ -0,0 +1,195 @@
+package com.eactive.eai.common.inflow.dual;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.springframework.context.ApplicationContext;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import com.eactive.eai.common.inflow.Bucket;
+import com.eactive.eai.common.inflow.InflowTargetVO;
+import com.eactive.eai.common.server.EAIServerManager;
+import com.eactive.eai.common.util.ApplicationContextProvider;
+
+/**
+ * DualRequestProcessor 단위 테스트.
+ *
+ * <p>RequestProcessor 클래스는 static 초기화 블록에서 EAIServerManager.getInstance()를 호출한다.
+ * @BeforeAll에서 mock ApplicationContext를 주입한 뒤 DualRequestProcessor를 최초 로딩시킨다.
+ *
+ * <p>주의: DualRequestProcessor를 필드 타입으로 선언하면 테스트 클래스 로딩 시 함께 로딩되어
+ * @BeforeAll 실행 전에 NPE가 발생한다. 반드시 메서드 본문 내에서만 참조해야 한다.
+ */
+class DualRequestProcessorTest {
+
+    @BeforeAll
+    static void setupMockApplicationContext() {
+        ApplicationContext mockContext = mock(ApplicationContext.class);
+        EAIServerManager mockServerManager = mock(EAIServerManager.class);
+        when(mockContext.getBean(EAIServerManager.class)).thenReturn(mockServerManager);
+        when(mockServerManager.getLocalServerName()).thenReturn("TEST-SERVER");
+        when(mockServerManager.getGroupInstId()).thenReturn("TEST-INST");
+
+        ReflectionTestUtils.setField(ApplicationContextProvider.class, "context", mockContext);
+    }
+
+    private InflowTargetVO makeVO(String name, long perSec, long threshold, String unit) {
+        InflowTargetVO vo = new InflowTargetVO();
+        vo.setName(name);
+        vo.setThresholdPerSecond(perSec);
+        vo.setThreshold(threshold);
+        vo.setThresholdTimeUnit(unit);
+        vo.setActivate(true);
+        return vo;
+    }
+
+    // -------------------------------------------------------------------------
+    // checkAdapterInflow
+    // -------------------------------------------------------------------------
+
+    @Test
+    void checkAdapterInflow_dualBucket_pass_returnsNull() {
+        DualRequestProcessor processor = new DualRequestProcessor();
+        DualBucket mockBucket = mock(DualBucket.class);
+        when(mockBucket.isAdapterPassDetail("ADAPTER_A")).thenReturn(null);
+
+        assertNull(processor.checkAdapterInflow(mockBucket, "ADAPTER_A"));
+        verify(mockBucket).isAdapterPassDetail("ADAPTER_A");
+    }
+
+    @Test
+    void checkAdapterInflow_dualBucket_blockedPerSecond() {
+        DualRequestProcessor processor = new DualRequestProcessor();
+        DualBucket mockBucket = mock(DualBucket.class);
+        when(mockBucket.isAdapterPassDetail("ADAPTER_A")).thenReturn(DualCustomBucket.RESULT_BLOCKED_PER_SECOND);
+
+        assertEquals(DualCustomBucket.RESULT_BLOCKED_PER_SECOND,
+                processor.checkAdapterInflow(mockBucket, "ADAPTER_A"));
+    }
+
+    @Test
+    void checkAdapterInflow_dualBucket_blockedThreshold() {
+        DualRequestProcessor processor = new DualRequestProcessor();
+        DualBucket mockBucket = mock(DualBucket.class);
+        when(mockBucket.isAdapterPassDetail("ADAPTER_A")).thenReturn(DualCustomBucket.RESULT_BLOCKED_THRESHOLD);
+
+        assertEquals(DualCustomBucket.RESULT_BLOCKED_THRESHOLD,
+                processor.checkAdapterInflow(mockBucket, "ADAPTER_A"));
+    }
+
+    @Test
+    void checkAdapterInflow_nonDualBucket_pass_returnsNull() {
+        DualRequestProcessor processor = new DualRequestProcessor();
+        Bucket mockBucket = mock(Bucket.class);
+        when(mockBucket.isAdapterPass("ADAPTER_A")).thenReturn(true);
+
+        assertNull(processor.checkAdapterInflow(mockBucket, "ADAPTER_A"));
+    }
+
+    @Test
+    void checkAdapterInflow_nonDualBucket_blocked_returnsBlocked() {
+        DualRequestProcessor processor = new DualRequestProcessor();
+        Bucket mockBucket = mock(Bucket.class);
+        when(mockBucket.isAdapterPass("ADAPTER_A")).thenReturn(false);
+
+        assertEquals("BLOCKED", processor.checkAdapterInflow(mockBucket, "ADAPTER_A"));
+    }
+
+    // -------------------------------------------------------------------------
+    // checkInterfaceInflow
+    // -------------------------------------------------------------------------
+
+    @Test
+    void checkInterfaceInflow_dualBucket_pass_returnsNull() {
+        DualRequestProcessor processor = new DualRequestProcessor();
+        DualBucket mockBucket = mock(DualBucket.class);
+        when(mockBucket.isInterfacePassDetail("IF_001")).thenReturn(null);
+
+        assertNull(processor.checkInterfaceInflow(mockBucket, "IF_001"));
+        verify(mockBucket).isInterfacePassDetail("IF_001");
+    }
+
+    @Test
+    void checkInterfaceInflow_dualBucket_blockedPerSecond() {
+        DualRequestProcessor processor = new DualRequestProcessor();
+        DualBucket mockBucket = mock(DualBucket.class);
+        when(mockBucket.isInterfacePassDetail("IF_001")).thenReturn(DualCustomBucket.RESULT_BLOCKED_PER_SECOND);
+
+        assertEquals(DualCustomBucket.RESULT_BLOCKED_PER_SECOND,
+                processor.checkInterfaceInflow(mockBucket, "IF_001"));
+    }
+
+    @Test
+    void checkInterfaceInflow_nonDualBucket_pass_returnsNull() {
+        DualRequestProcessor processor = new DualRequestProcessor();
+        Bucket mockBucket = mock(Bucket.class);
+        when(mockBucket.isInterfacePass("IF_001")).thenReturn(true);
+
+        assertNull(processor.checkInterfaceInflow(mockBucket, "IF_001"));
+    }
+
+    @Test
+    void checkInterfaceInflow_nonDualBucket_blocked_returnsBlocked() {
+        DualRequestProcessor processor = new DualRequestProcessor();
+        Bucket mockBucket = mock(Bucket.class);
+        when(mockBucket.isInterfacePass("IF_001")).thenReturn(false);
+
+        assertEquals("BLOCKED", processor.checkInterfaceInflow(mockBucket, "IF_001"));
+    }
+
+    // -------------------------------------------------------------------------
+    // buildInflowTargetErrorMsg
+    // -------------------------------------------------------------------------
+
+    @Test
+    void buildInflowTargetErrorMsg_perSecond_includesReqPerSec() {
+        DualRequestProcessor processor = new DualRequestProcessor();
+        InflowTargetVO vo = makeVO("결제API", 30, 1000, "MIN");
+
+        String msg = processor.buildInflowTargetErrorMsg(vo, DualCustomBucket.RESULT_BLOCKED_PER_SECOND);
+
+        assertEquals("API 호출 한도 초과 [결제API: 30req/sec]", msg);
+    }
+
+    @Test
+    void buildInflowTargetErrorMsg_threshold_includesReqPerUnit() {
+        DualRequestProcessor processor = new DualRequestProcessor();
+        InflowTargetVO vo = makeVO("결제API", 30, 1000, "MIN");
+
+        String msg = processor.buildInflowTargetErrorMsg(vo, DualCustomBucket.RESULT_BLOCKED_THRESHOLD);
+
+        assertEquals("API 호출 한도 초과 [결제API: 1000req/MIN]", msg);
+    }
+
+    @Test
+    void buildInflowTargetErrorMsg_blockedFallback_simpleFormat() {
+        DualRequestProcessor processor = new DualRequestProcessor();
+        InflowTargetVO vo = makeVO("결제API", 30, 1000, "MIN");
+
+        String msg = processor.buildInflowTargetErrorMsg(vo, "BLOCKED");
+
+        assertEquals("API 호출 한도 초과 [결제API]", msg);
+    }
+
+    @Test
+    void buildInflowTargetErrorMsg_nullBlockedType_simpleFormat() {
+        DualRequestProcessor processor = new DualRequestProcessor();
+        InflowTargetVO vo = makeVO("결제API", 30, 1000, "MIN");
+
+        String msg = processor.buildInflowTargetErrorMsg(vo, null);
+
+        assertEquals("API 호출 한도 초과 [결제API]", msg);
+    }
+
+    @Test
+    void buildInflowTargetErrorMsg_hourUnit() {
+        DualRequestProcessor processor = new DualRequestProcessor();
+        InflowTargetVO vo = makeVO("조회API", 10, 500, "HOUR");
+
+        String msg = processor.buildInflowTargetErrorMsg(vo, DualCustomBucket.RESULT_BLOCKED_THRESHOLD);
+
+        assertEquals("API 호출 한도 초과 [조회API: 500req/HOUR]", msg);
+    }
+}