daffa

Okay, acknowledging the constraints (no new `financeEventType`, reuse `SALE_FORWARD`/`SALE_REVERSAL`, minimal enrichment), and using the provided reference diff as a guide, here's the updated plan and corresponding code snippets for BEP.

**Core Decisions Based on Constraints:**

1.  **`financeEventType` Re-use:** Manual Adjustments will map to `SALE_FORWARD` (for PAY) or `SALE_REVERSAL` (for CANCEL).
2.  **New Request Contracts:** We'll define new Request classes in the `com.coupang.fp.contract.eats` project (separate repo) that *structurally resemble* existing sale requests but carry the specific manual adjustment data.
3.  **Simplified Enrichment:** The `EnrichmentManager` calls will be placeholders, simply wrapping the input DTO into an enriched DTO without external lookups.
4.  **Single Line Item:** The finance event will contain one line item representing the total adjustment.
5.  **Combined Actors/Orchestrators (following reference):** We won't create separate actors/orchestrators for PAY/CANCEL if the input DTO can distinguish between them.

**1. Define UpstreamEventNames (Add these to your `UpstreamEventName` enum):**

```java
public enum UpstreamEventName {
    // ... existing events ...
    MANUAL_ADJUSTMENTS_EATS_PAY("Manual Adjustments Eats Pay", "eats-merchant.settlement.manual.1"), // Topic for Eats adjustments
    MANUAL_ADJUSTMENTS_EATS_CANCEL("Manual Adjustments Eats Cancel", "eats-merchant.settlement.manual.1"), // Same topic, differentiate in actor
    MANUAL_ADJUSTMENTS_EATS_ADS_CANCEL("Manual Adjustments Eats Ads Cancel", "cmg_settlement_app.adjustment.eats.1"); // Topic for Ads adjustments

    // ... enum body ...
}
```

**2. Models (`models/internal/...`)**

* **Create DTOs:**

    ```java
    // models/internal/subscribers/src/main/java/com/coupang/fp/events/eats/EatsManualAdjustmentEventDTO.java
    package com.coupang.fp.events.eats;

    import com.coupang.fp.events.BaseEventDTO;
    import com.coupang.fp.events.UpstreamEventName;
    import lombok.Data;
    import jakarta.validation.constraints.NotNull;
    import java.math.BigDecimal;
    import java.time.LocalDate;

    @Data
    public class EatsManualAdjustmentEventDTO extends BaseEventDTO {
        @NotNull private String reconciliationKey;
        @NotNull private Long storeId;
        @NotNull private String manualAdjustmentId; // Renamed
        private String description;
        private String descriptionDetail;
        @NotNull private BigDecimal amount;
        @NotNull private BigDecimal supplyAmount;
        @NotNull private BigDecimal vat;
        @NotNull private String payTransactionType; // "PAY" or "CANCEL"
        @NotNull private String costType;
        private String expectedSettlementDate; // Keep as String for now if source sends String
        private String settlementType;
        private String settlementScheduleType;
        private Integer transferSchedule;
        private Integer schedule;
        private String currency; // Added
        private Long eventTime; // Added (epoch ms)
        private String orderId; // Optional

        @Override
        public UpstreamEventName getUpstreamEventName() {
            // Determine event name based on payTransactionType
            return "PAY".equalsIgnoreCase(this.payTransactionType) ?
                   UpstreamEventName.MANUAL_ADJUSTMENTS_EATS_PAY :
                   UpstreamEventName.MANUAL_ADJUSTMENTS_EATS_CANCEL;
        }
    }

    // models/internal/subscribers/src/main/java/com/coupang/fp/events/eats/EatsManualAdjustmentAdsEventDTO.java
    package com.coupang.fp.events.eats;

    import com.coupang.fp.events.BaseEventDTO;
    import com.coupang.fp.events.UpstreamEventName;
    import lombok.Data;
    import jakarta.validation.constraints.NotNull;
    import java.math.BigDecimal;
    import java.time.LocalDate;

    @Data
    public class EatsManualAdjustmentAdsEventDTO extends BaseEventDTO {
        @NotNull private String eatsAdjustmentId; // reconciliationKey for Ads
        @NotNull private Long paymentStoreId;
        @NotNull private BigDecimal amount;
        @NotNull private BigDecimal vat;
        @NotNull private Long approveTime; // Consider mapping to a date/time type if needed
        @NotNull private Long expectedRefundTime; // Settlement date (epoch ms)
        private String currency; // Added
        private Long eventTime; // Added (epoch ms)
        private String orderId; // Optional

        @Override
        public UpstreamEventName getUpstreamEventName() {
            // Ads Adjustments are always considered CANCEL from Coupang's perspective (money clawback)
            return UpstreamEventName.MANUAL_ADJUSTMENTS_EATS_ADS_CANCEL;
        }
    }
    ```

* **Create Enriched DTOs:**

    ```java
    // models/internal/enricher/src/main/java/com/coupang/fp/event/EatsManualAdjustmentEnrichedEvent.java
    package com.coupang.fp.event;

    import com.coupang.fp.events.BaseEventDTO;
    import com.coupang.fp.events.eats.EatsManualAdjustmentEventDTO;
    // Import StoreDetails if you decide to add store enrichment later
    import lombok.Getter;
    import lombok.NoArgsConstructor;
    import lombok.Setter;

    @Getter
    @NoArgsConstructor
    @Setter
    public class EatsManualAdjustmentEnrichedEvent extends BaseEventDTO {
        private EatsManualAdjustmentEventDTO eatsManualAdjustmentEventDTO;
        // private StoreDetails storeDetails; // Add if store enrichment is needed

        public EatsManualAdjustmentEnrichedEvent(EatsManualAdjustmentEventDTO eventDTO) {
            super(eventDTO.getUpstreamEventName(), eventDTO.getParentId(), eventDTO.getEventReceivedAt(), eventDTO.getEventCreatedAt(),
                    eventDTO.getEventId(), eventDTO.getCountryRegionCode(), eventDTO.getBusinessUnit(), eventDTO.getEventStatus(), eventDTO.getRunId());
            this.eatsManualAdjustmentEventDTO = eventDTO;
        }
        // public boolean isStoreDetailsDataAvailable() { return storeDetails != null; }
    }

    // models/internal/enricher/src/main/java/com/coupang/fp/event/EatsManualAdjustmentAdsEnrichedEvent.java
    package com.coupang.fp.event;

    import com.coupang.fp.events.BaseEventDTO;
    import com.coupang.fp.events.eats.EatsManualAdjustmentAdsEventDTO;
    // Import StoreDetails if you decide to add store enrichment later
    import lombok.Getter;
    import lombok.NoArgsConstructor;
    import lombok.Setter;

    @Getter
    @NoArgsConstructor
    @Setter
    public class EatsManualAdjustmentAdsEnrichedEvent extends BaseEventDTO {
        private EatsManualAdjustmentAdsEventDTO eatsManualAdjustmentAdsEventDTO;
        // private StoreDetails storeDetails; // Add if store enrichment is needed

        public EatsManualAdjustmentAdsEnrichedEvent(EatsManualAdjustmentAdsEventDTO eventDTO) {
            super(eventDTO.getUpstreamEventName(), eventDTO.getParentId(), eventDTO.getEventReceivedAt(), eventDTO.getEventCreatedAt(),
                    eventDTO.getEventId(), eventDTO.getCountryRegionCode(), eventDTO.getBusinessUnit(), eventDTO.getEventStatus(), eventDTO.getRunId());
            this.eatsManualAdjustmentAdsEventDTO = eventDTO;
        }
         // public boolean isStoreDetailsDataAvailable() { return storeDetails != null; }
    }
    ```

**3. ApiService (`service/bootstrap`)**

```java
// ApiService.java
// ... imports ...
import com.coupang.fp.events.eats.EatsManualAdjustmentEventDTO;
import com.coupang.fp.events.eats.EatsManualAdjustmentAdsEventDTO;

@Service
@Slf4j
@RequiredArgsConstructor
public class ApiService {
    // ... existing code ...

    public void processEvent(String topicName, String topicMessage) throws UnexpectedTopicNameException {
        BaseEventDTO dto = null;
        UpstreamEventName upstreamEventName = UpstreamEventName.fromTopic(topicName);
        log.info("Received message for Upstream Event Name: {}", upstreamEventName);

        switch (upstreamEventName) {
            // ... existing cases ...
            case MANUAL_ADJUSTMENTS_EATS_PAY: // Fall-through
            case MANUAL_ADJUSTMENTS_EATS_CANCEL: {
                dto = JsonConvertUtil.fromJson(topicMessage, EatsManualAdjustmentEventDTO.class);
                break;
            }
            case MANUAL_ADJUSTMENTS_EATS_ADS_CANCEL: {
                dto = JsonConvertUtil.fromJson(topicMessage, EatsManualAdjustmentAdsEventDTO.class);
                break;
            }
            default: {
                log.error("Received invalid topic name : {}", topicName);
                throw new UnexpectedTopicNameException(topicName + "\n" + topicMessage);
            }
        }
        // ... rest of the method ...
    }
}
```

**4. Enrichment (`service/enricher`)**

* **`EnrichmentManager`:**

    ```java
    // EnrichmentManager.java
    // ... imports ...
    import com.coupang.fp.events.eats.EatsManualAdjustmentEventDTO;
    import com.coupang.fp.events.eats.EatsManualAdjustmentAdsEventDTO;
    import com.coupang.fp.event.EatsManualAdjustmentEnrichedEvent;
    import com.coupang.fp.event.EatsManualAdjustmentAdsEnrichedEvent;

    @Component
    @Slf4j
    public class EnrichmentManager {
        // ... existing code ...
        @PostConstruct
        public void init() {
            // ... existing mappings ...
            eventNameToEnrichedEventDTOMap.put(MANUAL_ADJUSTMENTS_EATS_PAY, EatsManualAdjustmentEventDTO.class);
            eventNameToEnrichedEventDTOMap.put(MANUAL_ADJUSTMENTS_EATS_CANCEL, EatsManualAdjustmentEventDTO.class);
            eventNameToEnrichedEventDTOMap.put(MANUAL_ADJUSTMENTS_EATS_ADS_CANCEL, EatsManualAdjustmentAdsEventDTO.class);
        }

        // Placeholder - No actual enrichment
        public EatsManualAdjustmentEnrichedEvent enrichEatsManualAdjustmentEvent(BaseEventDTO baseEventDTO) {
            validateEventAndDtoType(baseEventDTO); // Create or reuse a validation helper
            log.info("Placeholder enrichment for event id : {}", baseEventDTO.getEventId());
            return new EatsManualAdjustmentEnrichedEvent((EatsManualAdjustmentEventDTO) baseEventDTO);
        }

        // Placeholder - No actual enrichment
        public EatsManualAdjustmentAdsEnrichedEvent enrichEatsManualAdjustmentAdsEvent(BaseEventDTO baseEventDTO) {
             validateEventAndDtoType(baseEventDTO); // Create or reuse a validation helper
            log.info("Placeholder enrichment for event id : {}", baseEventDTO.getEventId());
            return new EatsManualAdjustmentAdsEnrichedEvent((EatsManualAdjustmentAdsEventDTO) baseEventDTO);
        }

        private void validateEventAndDtoType(BaseEventDTO baseEventDTO) {
         Type expectedClassType = eventNameToEnrichedEventDTOMap.get(baseEventDTO.getUpstreamEventName());
         if (!(baseEventDTO.getClass().equals(expectedClassType)))
            throw new IllegalArgumentException("This enrichment expects an input of type " + expectedClassType.getTypeName());
       }
    }
    ```
* **No new Flow or Template classes needed** as enrichment is a placeholder.

**5. ID Generators (`service/event-registry`)**

```java
// EntityIdGenerator.java
private String getEntityIdForEvent(BaseEventDTO baseEventDTO) {
    switch (baseEventDTO.getUpstreamEventName()) {
        // ... existing cases ...
        case MANUAL_ADJUSTMENTS_EATS_PAY, MANUAL_ADJUSTMENTS_EATS_CANCEL -> {
             return ENTITY_ID_PREFIX + SEPARATOR_HYPHEN + baseEventDTO.getCountryRegionCode() + SEPARATOR_HYPHEN + ((EatsManualAdjustmentEventDTO)baseEventDTO).getReconciliationKey();
        }
        case MANUAL_ADJUSTMENTS_EATS_ADS_CANCEL -> {
             return ENTITY_ID_PREFIX + SEPARATOR_HYPHEN + baseEventDTO.getCountryRegionCode() + SEPARATOR_HYPHEN + ((EatsManualAdjustmentAdsEventDTO)baseEventDTO).getEatsAdjustmentId(); // Use eatsAdjustmentId for Ads
        }
    }
    // ...
}

// EventIdGenerator.java
public String getEventId(BaseEventDTO baseEventDTO) {
    switch (baseEventDTO.getUpstreamEventName()) {
        // ... existing cases ...
         case MANUAL_ADJUSTMENTS_EATS_PAY, MANUAL_ADJUSTMENTS_EATS_CANCEL -> {
             return EVENT_ID_PREFIX + SEPARATOR_HYPHEN + baseEventDTO.getCountryRegionCode() + SEPARATOR_HYPHEN + ((EatsManualAdjustmentEventDTO)baseEventDTO).getReconciliationKey();
        }
        case MANUAL_ADJUSTMENTS_EATS_ADS_CANCEL -> {
             return EVENT_ID_PREFIX + SEPARATOR_HYPHEN + baseEventDTO.getCountryRegionCode() + SEPARATOR_HYPHEN + ((EatsManualAdjustmentAdsEventDTO)baseEventDTO).getEatsAdjustmentId(); // Use eatsAdjustmentId for Ads
        }
    }
     // ...
}
```

**6. Finance Event Publisher (`service/finance-event-publisher`)**

* **`EventPublisherFacade`:**

    ```java
    // EventPublisherFacade.java
    // ... imports ...
    import com.coupang.fp.event.EatsManualAdjustmentEnrichedEvent;
    import com.coupang.fp.event.EatsManualAdjustmentAdsEnrichedEvent;

    @Component
    @Slf4j
    @RequiredArgsConstructor
    public class EventPublisherFacade {
        // ... existing code ...

        public boolean publish(EatsManualAdjustmentEnrichedEvent event) throws PublisherException {
            eventEventRelayer.relay(Collections.singletonList(event));
            return true;
        }

        public boolean publish(EatsManualAdjustmentAdsEnrichedEvent event) throws PublisherException {
            eventEventRelayer.relay(Collections.singletonList(event));
            return true;
        }
    }
    ```

* **Finance Event Contracts (Separate Repo - `com.coupang.fp.contract.eats`):**
    * Define these new classes:
        * `EatsManualAdjustmentSaleForwardRequest`
        * `EatsManualAdjustmentSaleReversalRequest`
        * `EatsManualAdjustmentAdsSaleReversalRequest`
        * `EatsManualAdjustmentTransactionContext`
        * `EatsManualAdjustmentLineItemContext`
        * `EatsManualAdjustmentAdsTransactionContext`
        * `EatsManualAdjustmentAdsLineItemContext`
    * *(These will structurally mirror existing sale/reversal requests but contain fields specific to manual adjustments, like `costType`)*.

* **`EnrichedEventToFinanceEventMapper`:**

    ```java
    // EnrichedEventToFinanceEventMapper.java
    // ... imports ...
    import com.coupang.fp.contract.eats.*; // Import the new contract classes

    @Component
    @Slf4j
    public class EnrichedEventToFinanceEventMapper {
        // ... existing code ...

        private void mapEatsEvent(Object result, Map<FinanceEventType, FinanceBaseEvent<?, ?>> financeEventMap) {
            if (result instanceof EatsSaleForwardRequest) {
                financeEventMap.put(FinanceEventType.SALE_FORWARD, (EatsSaleForwardRequest) result);
            } else if (result instanceof EatsSaleReverseRequest) {
                financeEventMap.put(FinanceEventType.SALE_REVERSAL, (EatsSaleReverseRequest) result);
            } else if (result instanceof EatsAdsSaleForwardRequest) { // Handle Ads (if you had a PAY case)
                financeEventMap.put(FinanceEventType.SALE_FORWARD, (EatsAdsSaleForwardRequest) result);
            } else if (result instanceof EatsAdsSaleReverseRequest) { // Handle Ads Cancel
                financeEventMap.put(FinanceEventType.SALE_REVERSAL, (EatsAdsSaleReverseRequest) result);
            }
             // *** Add Mappings for New Manual Adjustment Requests ***
            else if (result instanceof EatsManualAdjustmentSaleForwardRequest) {
                financeEventMap.put(FinanceEventType.SALE_FORWARD, (EatsManualAdjustmentSaleForwardRequest) result);
            } else if (result instanceof EatsManualAdjustmentSaleReversalRequest) {
                 financeEventMap.put(FinanceEventType.SALE_REVERSAL, (EatsManualAdjustmentSaleReversalRequest) result);
            } else if (result instanceof EatsManualAdjustmentAdsSaleReversalRequest) {
                 financeEventMap.put(FinanceEventType.SALE_REVERSAL, (EatsManualAdjustmentAdsSaleReversalRequest) result);
            }
            // *** End New Mappings ***
            else {
                throw new IllegalArgumentException("Unsupported event type: " + result.getClass());
            }
        }
    }
    ```

* **`MapperRegistry`:**

    ```java
    // MapperRegistry.java
    // ... imports ...
    import com.coupang.fp.mapper.mapper_registry.eats.EatsManualAdjustmentEnrichedEventToSaleForwardRequestMapperImpl;
    import com.coupang.fp.mapper.mapper_registry.eats.EatsManualAdjustmentEnrichedEventToSaleReversalRequestMapperImpl;
    import com.coupang.fp.mapper.mapper_registry.eats.EatsManualAdjustmentAdsEnrichedEventToSaleReversalRequestMapperImpl;

    @Slf4j
    public class MapperRegistry {
        // ... existing code ...

        private static List<BaseEnrichedEventMapping<? extends BaseEventDTO, ?>> getEatsEventMappers(EnrichedEventStepInput<? extends BaseEventDTO> enrichedEventStepInput) {
            UpstreamEventName eventName = enrichedEventStepInput.getEvent().getUpstreamEventName();
            return switch (eventName) {
                case EATS_ORDER -> List.of(new EatsOrderEnrichedEventToSaleForwardEventMappingImpl());
                case EATS_CANCEL -> List.of(new EatsOrderEnrichedEventToSaleReversalEventMappingImpl());
                case EATS_ADS_ORDER -> List.of(new EatsAdsEnrichedEventToSaleForwardEventMappingImpl()); // If Ads PAY exists
                case EATS_ADS_CANCEL -> List.of(new EatsAdsEnrichedEventToSaleReversalEventMappingImpl());
                // *** Add Mappings for Manual Adjustments ***
                case MANUAL_ADJUSTMENTS_EATS_PAY -> List.of(new EatsManualAdjustmentEnrichedEventToSaleForwardRequestMapperImpl());
                case MANUAL_ADJUSTMENTS_EATS_CANCEL -> List.of(new EatsManualAdjustmentEnrichedEventToSaleReversalRequestMapperImpl());
                case MANUAL_ADJUSTMENTS_EATS_ADS_CANCEL -> List.of(new EatsManualAdjustmentAdsEnrichedEventToSaleReversalRequestMapperImpl());
                // *** End New Mappings ***
                default -> new ArrayList<>();
            };
        }
    }
    ```

* **Mapper Implementations (MapStruct):**
    * Create the necessary MapStruct interfaces (`EatsManualAdjustmentEnrichedEventToSaleForwardRequestMapper`, etc.).
    * Implement the mapping logic:
        * Set `financeEventType` to `SALE_FORWARD` or `SALE_REVERSAL`.
        * Map relevant fields from `EatsManualAdjustmentEventDTO` or `EatsManualAdjustmentAdsEventDTO` to the corresponding `TransactionContext` and `LineItemContext`.
        * Use `EatsMapperUtils` (potentially extended) to create the single `LineItem`.

    ```java
    // Example: EatsManualAdjustmentEnrichedEventToSaleForwardRequestMapper.java
    package com.coupang.fp.mapper.mapper_registry.eats;

    import com.coupang.fp.contract.eats.*; // Import the new contract classes
    import com.coupang.fp.event.EatsManualAdjustmentEnrichedEvent;
    import com.coupang.fp.helper.DateTimeHelper;
    import com.coupang.fp.mapper.mapper_registry.common.BaseEnrichedEventMapping;
    import org.mapstruct.Mapper;
    import org.mapstruct.Mapping;
    import org.mapstruct.Named;
    // ... other imports ...

    @Mapper(componentModel = "spring", builder = @Builder(disableBuilder = true))
    public interface EatsManualAdjustmentEnrichedEventToSaleForwardRequestMapper
        extends BaseEnrichedEventMapping<EatsManualAdjustmentEnrichedEvent, EatsManualAdjustmentSaleForwardRequest> {

        @Mapping(target = "financeEventType", expression = "java(com.coupang.fp.contract.enums.FinanceEventType.SALE_FORWARD)") // Use SALE_FORWARD
        @Mapping(target = "businessUnit", expression = "java(com.coupang.fp.enums.BusinessUnit.EATS)")
        // ... Map common fields like storeId, currency, etc. ...
        @Mapping(target = "referenceId", source = "eatsManualAdjustmentEventDTO.manualAdjustmentId")
        @Mapping(target = "referenceIdType", constant = "MANUAL_ADJUSTMENT_ID")
        @Mapping(target = "transactionTimestamp", source = "eatsManualAdjustmentEventDTO.eventTime", qualifiedByName = "epochToZonedDateTime")
        @Mapping(target = "realizationTimestamp", expression = "java(java.time.ZonedDateTime.now())")
        @Mapping(target = "lineItems", source = "eatsManualAdjustmentEnrichedEvent", qualifiedByName = "mapManualAdjLineItem")
        // Map fields specific to ManualAdjustmentTransactionContext
        @Mapping(target = "transactionContext.costType", source = "eatsManualAdjustmentEventDTO.costType")
        @Mapping(target = "transactionContext.payTransactionType", source = "eatsManualAdjustmentEventDTO.payTransactionType")
        // ... map other context fields ...

        EatsManualAdjustmentSaleForwardRequest map(EatsManualAdjustmentEnrichedEvent eatsManualAdjustmentEnrichedEvent);

        @Named("epochToZonedDateTime")
        default ZonedDateTime epochToZonedDateTime(Long epochMillis) {
            return DateTimeHelper.fromEpochMilliToZonedDateTime(epochMillis);
        }

        @Named("mapManualAdjLineItem")
        default List<LineItem<EatsManualAdjustmentLineItemContext>> mapManualAdjLineItem(EatsManualAdjustmentEnrichedEvent event) {
           return EatsMapperUtils.extractEatsManualAdjustmentLineItems(event); // Need to create this method
        }
    }
    ```

    * Create similar mappers for `EatsManualAdjustmentEnrichedEventToSaleReversalRequestMapper` and `EatsManualAdjustmentAdsEnrichedEventToSaleReversalRequestMapper`.

**7. Orchestrator (`service/orchestrator`)**

```java
// EventDispatcher.java
// ... imports ...
import com.coupang.fp.events.eats.EatsManualAdjustmentEventDTO;
import com.coupang.fp.events.eats.EatsManualAdjustmentAdsEventDTO;

@Component
@RequiredArgsConstructor
@Slf4j
public class EventDispatcher {
    // ... inject new orchestrators ...
    private final EatsManualAdjustmentEventOrchestrator eatsManualAdjustmentEventOrchestrator;
    private final EatsManualAdjustmentAdsEventOrchestrator eatsManualAdjustmentAdsEventOrchestrator;

    // ... existing code ...

    private void handleEvents(BaseEventDTO baseEventDTO) {
        // ... existing cases ...
        case MANUAL_ADJUSTMENTS_EATS_PAY: // Fall-through
        case MANUAL_ADJUSTMENTS_EATS_CANCEL: {
            if (baseEventDTO instanceof EatsManualAdjustmentEventDTO eventDTO) {
                eatsManualAdjustmentEventOrchestrator.processEvent(eventDTO);
            }
            break;
        }
        case MANUAL_ADJUSTMENTS_EATS_ADS_CANCEL: {
            if (baseEventDTO instanceof EatsManualAdjustmentAdsEventDTO eventDTO) {
                eatsManualAdjustmentAdsEventOrchestrator.processEvent(eventDTO);
            }
            break;
        }
        // ... default case ...
    }
}

// Create new orchestrator classes:
// EatsManualAdjustmentEventOrchestrator.java
// EatsManualAdjustmentAdsEventOrchestrator.java
// (These will be very similar to EatsAdsOrderEventOrchestrator, just calling the corresponding enrich and publish methods)
```

**8. Subscribers (`service/subscribers`)**

```java
// CounectConsumerConfig.java
@Configuration
@RequiredArgsConstructor
@Slf4j
public class CounectConsumerConfig {
    // ... inject new actors ...
    @Autowired private EatsManualAdjustmentEventActor eatsManualAdjustmentEventActor;
    @Autowired private EatsManualAdjustmentAdsEventActor eatsManualAdjustmentAdsEventActor;

    @PostConstruct
    public void initialize() {
        // ... existing subscriptions ...
        subscriberForEatsManualAdjustmentKafkaEvent();
        subscriberForEatsManualAdjustmentAdsKafkaEvent();
    }

    public void subscriberForEatsManualAdjustmentKafkaEvent() {
        String topic = "eats-merchant.settlement.manual.1"; // Topic for Eats adjustments
        if (!topicProperties.getConsumers().containsKey(topic)) { /*...*/ return; }
        String userId = topicProperties.getConsumers().get(topic).userId();
        Consumer<byte[]> consumer = counectFactory.createConsumer(counectContext, topic, userId, byte[].class);
        consumer.subscribe(DispatcherMode.PARALLEL, this.eatsManualAdjustmentEventActor);
        log.info("subscriberForEatsManualAdjustmentKafkaEvent:{} has subscribed successfully", topic);
        consumerList.add(consumer);
    }

     public void subscriberForEatsManualAdjustmentAdsKafkaEvent() {
        String topic = "cmg_settlement_app.adjustment.eats.1"; // Topic for Ads adjustments
        if (!topicProperties.getConsumers().containsKey(topic)) { /*...*/ return; }
        String userId = topicProperties.getConsumers().get(topic).userId();
        Consumer<byte[]> consumer = counectFactory.createConsumer(counectContext, topic, userId, byte[].class);
        consumer.subscribe(DispatcherMode.PARALLEL, this.eatsManualAdjustmentAdsEventActor);
        log.info("subscriberForEatsManualAdjustmentAdsKafkaEvent:{} has subscribed successfully", topic);
        consumerList.add(consumer);
    }
    // ... preDestroy ...
}

// Create new Actor classes:
// EatsManualAdjustmentEventActor.java
// EatsManualAdjustmentAdsEventActor.java
// (These will be similar to EatsAdsOrderEventActor, deserializing to the correct DTO and calling eventDispatcher)
// Note: EatsManualAdjustmentEventActor will handle BOTH PAY and CANCEL from the same topic.
```

**9. Unit Tests**

Create corresponding unit tests for all new classes and modify tests for changed classes.

This detailed implementation plan follows the requested constraints and the reference example. Remember to define the new Contract classes in the separate repository.