검색을 위한 데이터 다루기

// 가게 이벤트 수신
@SqsListener(value = "shopSqs", deletionPolicy = NEVER)
public void tagsUpdateListen(@Payload ShopChangeEvent shopChangeEvent, Acknowledgment acknowledgment) {
    shopElasticsearch.update(createUpdateRequest(shopChangeEvent));
}
...
// 리뷰 이벤트 수신
@SqsListener(value = "reviewSqs", deletionPolicy = NEVER)
public void tagsUpdateListen(@Payload ReviewChangeEvent reviewChangeEvent, Acknowledgment acknowledgment) {
    shopElasticsearch.update(createUpdateRequest(reviewChangeEvent));
}

// BulkRequest (Buffer 역할을 하는 객체)
public class BulkRequest extends ActionRequest implements CompositeIndicesRequest, WriteRequest<BulkRequest> {

    private static final int REQUEST_OVERHEAD = 50;

    /**
     * Requests that are part of this request. It is only possible to add things that are both {@link ActionRequest}s and
     * {@link WriteRequest}s to this but java doesn't support syntax to declare that everything in the array has both types so we declare
     * the one with the least casts.
     */
    final List<DocWriteRequest<?>> requests = new ArrayList<>();
    ..
    ..
    ..

    /**
     * Adds a list of requests to be executed. Either index or delete requests.
     */
    public BulkRequest add(DocWriteRequest<?>... requests) {
        for (DocWriteRequest<?> request : requests) {
            add(request, null);
        }
        return this;
    }
}

// BufferProcessor add메소드를 통해 BulkRequest에 request를 만들어서 담는 부분
bulkProcessor.add(new IndexRequest("twitter", "_doc", "1").source(/* your doc here */));
bulkProcessor.add(new DeleteRequest("twitter", "_doc", "2"));

// BulkProcessor Handler
public void execute(BulkRequest bulkRequest, long executionId) {
    Runnable toRelease = () -> {};
    boolean bulkRequestSetupSuccessful = false;
    try {
        listener.beforeBulk(executionId, bulkRequest);
        semaphore.acquire();
        toRelease = semaphore::release;
        CountDownLatch latch = new CountDownLatch(1);
        retry.withBackoff(consumer, bulkRequest, new ActionListener<BulkResponse>() {
            @Override
            public void onResponse(BulkResponse response) {
                try {
                    listener.afterBulk(executionId, bulkRequest, response);
                } finally {
                    semaphore.release();
                    latch.countDown();
                }
            }

            @Override
            public void onFailure(Exception e) {
                try {
                    listener.afterBulk(executionId, bulkRequest, e);
                } finally {
                    semaphore.release();
                    latch.countDown();
                }
            }
        });
        bulkRequestSetupSuccessful = true;
        if (concurrentRequests == 0) {
            latch.await();
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        logger.info(() -> new ParameterizedMessage("Bulk request {} has been cancelled.", executionId), e);
        listener.afterBulk(executionId, bulkRequest, e);
    } catch (Exception e) {
        logger.warn(() -> new ParameterizedMessage("Failed to execute bulk request {}.", executionId), e);
        listener.afterBulk(executionId, bulkRequest, e);
    } finally {
        if (bulkRequestSetupSuccessful == false) {  // if we fail on client.bulk() release the semaphore
            toRelease.run();
        }
    }
}

# 실제 코드는 더 복잡하여 예시 코드를 기재합니다.

/**
 * data는 유연하게 받을 수 있도록 제네릭을 사용하였고, 실제 검색에서는 Shop객체를 받고 있다.
 * acknowledgments는 수신된 이벤트들에 대한 정상 처리 여부에 따라 sqs에 ack 처리를 하기 위해 보관  
*/
@Getter
@Setter
@NoArgsConstructor
public class BulkData<T extends IndexDomain> {
    private T data;
    private Set<Acknowledgment> acknowledgments;

    @Builder
    public BulkData(T data, Set<Acknowledgment> acknowledgments) {
        this.data = data;
        this.acknowledgments = acknowledgments;
    }
}

@Getter
@Setter
public class Shop extends IndexDomain {
    private Long shopNumber;
    private List<Menu> menus;
    private String content;
    private String subject;
    
    @Builder
    public Shop(Long shopNumber, List<Menu> menus, String content, String subject) {
        this.shopNumber = shopNumber;
        this.menus = menus;
        this.content = content;
        this.subject = subject;
    }
}

// 이벤트 변경정보를 가지는 객체의 공통 추상 클래스
public abstract class ShopUpdatableDomain extends UpdatableDomain<Shop> {

    @Override
    public Shop convert() {
        Shop shop = Shop.builder().build();
        convert(shop);
        return shop;
    }
}

// 메뉴 변경정보를 가지고 있는 ShopMenu
@Getter
@Setter
@NoArguments
public class ShopMenu extends ShopUpdatableDomain {
    private Long shopNumber;
    private List<Menu> menus;
    
    @Builder
    public ShopMenu(Long shopNumber, List<Menu> menus) {
        this.shopNumber = shopNumber;
        this.menus = menus;
    }

    @Override
    public void convert(Shop shop) {
        shop.setShopNumber(shopNumber);
        shop.setMenus(menus);
    }
}

// 리뷰 변경정보를 가지고 있는 ShopReview
@Getter
@Setter
@NoArguments
public class ShopReview extends ShopUpdatableDomain {
    private Long shopNumber;
    private String content;
    private String subject;

    @Builder
    public ShopReview(Long shopNumber, String content, String subject) {
        this.shopNumber = shopNumber;
        this.content = content;
        this.subject = subject;
    }

    @Override
    public void convert(Shop shop) {
        shop.setShopNumber(shopNumber);
        shop.setContent(content);
        shop.setSubject(subject);
    }
}

// 이벤트 수신과 Bulk processor에 이벤트 내용 전달
@SqsListener(value = "menu-sqs", deletionPolicy = NEVER)
public void shopMenuV3UpdateListen(@Payload ShopMenuEvent eventDto, Acknowledgment acknowledgment) {
    bulkProcessor.add(BulkProcessorBucket.<ShopMenu>builder()
        .data(ShopMenu.builder()
            .shopNumber(eventDto.getShopId())
            .menus(eventDto.getMenus())
        .build())
    .build());
}
...
@SqsListener(value = "reviewSqs", deletionPolicy = NEVER)
public void tagsUpdateListen(@Payload ReviewChangeEvent reviewChangeEvent, Acknowledgment acknowledgment) {
    bulkProcessor.add(BulkProcessorBucket.<ShopReview>builder()
        .data(ShopReview.builder()
            .shopNumber(eventDto.getShopId())
            .content(eventDto.getContent())
            .subject(eventDto.getSubject())
        .build())
    .build());
}


// bulk processor add 코드
@Slf4j
public class BulkProcessor<S extends IndexDomain> implements Closeable {

    private Map<String, BulkData<S>> datas;
    ...
    ..
    ...
    private synchronized void addData(BulkProcessorBucket request) {
        String id = request.getData().idValue();

        // bulk buffer안에 동일한 가게가 존재하는지 조회
        if (this.datas.containsKey(id)) {
            BulkData<S> bulkData = this.datas.get(id);
            
            // 들어온 이벤트 데이터를 기존 shop 변경분 정보의 매핑
            request.getData().convert(bulkData.getData());
            return;
        }
        this.datas.put(id, BulkData.<S>builder()
            .data((S) request.getData().convert())
            .build());
    }
    ...
    ..
    .
}