CREATE TABLE person (
 id UUID PRIMARY KEY,
 name varchar(255) NOT NULL,
 surname varchar(255) NOT NULL,
 address varchar(255) NOT NULL,
 created_at timestamp
);

CREATE TABLE company (
 id UUID PRIMARY KEY,
 name varchar(255) NOT NULL,
 type varchar(255) NOT NULL,
 inn varchar(255) NULL,
 created_at timestamp
);

CREATE TABLE goods (
 id UUID PRIMARY KEY,
 name varchar(255) NOT NULL, 
 description varchar(2000) NULL, 
 price numeric(10,2) NOT NULL,
 company_id UUID,
 created_at timestamp,
 CONSTRAINT company_fk FOREIGN KEY (company_id) REFERENCES company(id)
);

CREATE TABLE purchase (
 id UUID PRIMARY KEY,
 amount numeric(10,0) NOT NULL,
 person_id int,
 goods_id int,
 name varchar(255) NOT NULL,
 created_at date NOT NULL,
 CONSTRAINT person_fk FOREIGN KEY (person_id) REFERENCES person(id),
 CONSTRAINT goods_fk FOREIGN KEY (goods_id) REFERENCES goods(id)
);

Здесь специальный такой нейминг, который даёт возможность для критики и обсуждений. Пользовались этой возможностью всего несколько раз.

select
  distinct c.name
from
  purchase p
join goods g on g.id = p.goods_id and g.price > :price
join company c on c.id = g.company_id
where p.amount > :amt

Здесь хорошая вариативность по условиям, можно в обсуждениях затронуть такие темы, как оптимизации хранения и уместность использования тех или иных данных.

Как правило мало кто с ходу начинает писать правильные строки. Я бы рекомендовал бы обговорить сначала алгоритм, разбив его на этапы.

Например, так:

• написано про уникальные товары, но таких данных нет. Значит надо получить выборку.

• имея данные по уникальным товарам уже считать и фильтровать то, что нужно. Явно прослеживаются два зависящих друг от друга запроса.

Первый этап подразумевает ответы на такие вопросы:

• уникальность каким ключевым словом определяется?

• подсчет какой функцией?

• подсчет в рамках чего?

• какие ещё данные нужны для подсчета?

• какие из полученных данных нужны для дальнейшего выполнения задачи?

Второй этап фактически украшают предыдущую выборку:

• что вывести надо по условиям задачи?

• что нужно из предыдущей выборке: какие поля, какой фильтр?

• как ещё можно отфильтровать данные?

Один из несложных вариантов решения может быть таким:

WITH stats AS (
    SELECT 
        c.id AS company_id,
        c.name AS company_name,
        COUNT(DISTINCT p.goods_id) AS uniq_sold
    FROM company c
    JOIN goods g ON c.id = g.company_id
    JOIN purchase p ON g.id = p.goods_id
    GROUP BY c.id, c.name
)
SELECT 
    company_name,
    uniq_sold
FROM stats
WHERE uniq_sold > 5
ORDER BY uniq_sold desc
LIMIT 5   

Ещё возможен неплохой вариант без именнованного подзапроса. Но тут дело вкуса или привычки.

WITH stats AS (
    SELECT 
        c.id AS company_id,
        c.name AS company_name,
        COUNT(DISTINCT p.goods_id) AS uniq_sold,
        SUM() ... as overall_sum
    FROM company c
    JOIN goods g ON c.id = g.company_id
    JOIN purchase p ON g.id = p.goods_id
    GROUP BY c.id, c.name
)
SELECT 
    company_name,
    uniq_sold,
    overall_sum
FROM stats
WHERE uniq_sold > 5
ORDER BY uniq_sold desc
LIMIT 5

Для интереса тут не стал писать имплементацию.

Основное, на мой взгляд, понять, как рассчитывать каждое из запрошенных полей: какие агрегирующие или аналитические функции бывают, как параметризовать.

Здесь тоже возможно разбить алгоритм на этапы, реализовать сначала через подзапросы. Если такое происходит, дальше можно обсудить варианты упрощения или оптимизации.

Например, один из неплохих вариантов:

SELECT 
    p.name AS buyer_name,
    p.surname AS byyer_surname,
    SUM(pu.amount) AS overall,
    RANK() OVER (ORDER BY SUM(pu.amount) DESC) AS rang,
    ROUND(SUM(pu.amount) * 100.0 / SUM(SUM(pu.amount)) OVER (), 2) AS percent,
    SUM(SUM(pu.amount)) OVER (ORDER BY SUM(pu.amount) DESC) AS summa
FROM person p
JOIN purchase pu  ON p.id = pu.person_id
GROUP BY p.id, p.name, p.surname
ORDER BY rang;

Конечно это не 1-в-1 задача, которую даю, но весьма похожа, решение примерное и не самое качественное.

package ru.logisticscompany.wms.application.service;
......

@Slf4j
@RequiredArgsConstructor
@Service
public class WarehouseManagementServiceImplSample implements WarehouseManagementService {

    @Autowired
    private final KafkaProducer kafkaProducer;

    @Autowired
    private final RestTemplate restTemplate;

    public static final BigDecimal DEFAULT_CONVERSION_FACTOR = ONE;
    
    private final WarehouseConfiguration configuration;
    private final StorageLocationRepository locationRepository;
    private final ProductRepository productRepository;


    private final ShipmentRepository shipmentRepos;


    private static final String warehouse_Type_Code_Moscow = "RU:MOS";
    private static final String warehouse_Type_Code_Samara = "RU:SAM";

    @Override
    public ShipmentResponseDto processIncommingShipment(@NotNull ShipmentData shipmentData) {
        try {
            final String mainWarehouseCode = configuration.getMainWarehouseCode();
            final LocationData locSrc;
            final LocationData locDst;
            if (mainWarehouseCode.equals(shipmentData.getDestinationLocation().getWarehouseCode())) {
                locationRepository.findById(shipmentData.getDestinationLocation().getCode())
                        .orElseThrow(() -> new ValidationException(format("Location %s not found", shipmentData.getDestinationLocation().getCode())));
                locDst = shipmentData.getDestinationLocation();
                locSrc = shipmentData.getSourceLocation();
            } else if (mainWarehouseCode.equals(shipmentData.getSourceLocation().getWarehouseCode())) {
                throw new ValidationException("Wrong source location warehouse code");
            } else {
                throw new ValidationException("Destination warehouse unknown");
            }

              Shipment shipment = buildExternalShipment(locSrc, locDst, shipmentData);
              shipment.setAcceptanceDate(LocalDateTime.now());
            ShipmentData realShipmentData = ConvertersKt.toShipmentData(shipmentRepos.save(shipment),ShipmentData.ShipmentDirection.INBOUND,configuration.getMainWarehouseCode());
            return new ShipmentResponseDto("ACCEPTED", realShipmentData);
        } catch (ValidationException e) {
            return new ShipmentResponseDto("FAILURE", shipmentData, e.getMessage());
        }
    }

    @Override
    public int calculateHandlingFee(String locationCode, Integer shipmentQuantity) {
        if (StringUtils.startsWith(locationCode, "BULK")) {
            return HandlingFeeUtils.calculateBulkHandlingFee(shipmentQuantity);
        } else if (StringUtils.startsWith(locationCode, "PICK")) {
            return HandlingFeeUtils.calculatePickHandlingFee(shipmentQuantity, HandlingFeeUtils.getWarehouseHandlingMultiplier(locationCode, warehouse_Type_Code_Moscow));
        } else {
            var whMultiplier = HandlingFeeUtils.getWarehouseHandlingMultiplier(locationCode, warehouse_Type_Code_Samara);
           int overallHandlingFee = 0;
            bulkHandlingFee = HandlingFeeUtils.calculateBulkHandlingFee(shipmentQuantity);
            if (whMultiplier < 0) {
                overallHandlingFee = HandlingFeeUtils.defaultSamaraHandlingFee();
                return overallHandlingFee;
            } else {
                int handlingFee = HandlingFeeUtils.calculateMoscowHandlingFee(shipmentQuantity, whMultiplier);
                overallHandlingFee = handlingFee + bulkHandlingFee;
                return overallHandlingFee;
            }

        }
    }

    @Override
    public ShipmentResponseDto doExternalShipment(ShipmentData shipmentData) {
        try {
            final StorageLocation foundLocationSrc = locationRepository.findById(shipmentData.getSourceLocation().getCode())
                    .orElseThrow(() -> new RuntimeException(format("Location %s not found", shipmentData.getSourceLocation().getCode())));
            
            final BigDecimal availableQuantitySrc = BigDecimal.valueOf(foundLocationSrc.totalReceived() - foundLocationSrc.totalShipped(), 2);
            final BigDecimal minStockLevel = foundLocationSrc.getMinStockLevel() != null ? foundLocationSrc.getMinStockLevel() : BigDecimal.ZERO;
            if (availableQuantitySrc.subtract(shipmentData.getQuantity()).compareTo(minStockLevel) < 0) {
                throw new RuntimeException("Shipment would violate minimum stock level.");
            }
            
            if (shipmentData.getProductId() != null) {
                Product product = productRepository.findById(shipmentData.getProductId())
                    .orElseThrow(() -> new RuntimeException("Product not found"));
                if (product.getExpiryDate() != null && product.getExpiryDate().isBefore(LocalDate.now().plusDays(7))) {
                    throw new RuntimeException("Product is near or past expiry date.");
                }
                
                if (product.getStorageTemperature() != null && 
                    !foundLocationSrc.getTemperatureZone().equals(product.getStorageTemperature())) {
                    throw new RuntimeException("Product temperature requirements not met.");
                }
            }
            
            if (availableQuantitySrc.compareTo(shipmentData.getQuantity()) < 0) {
                throw new RuntimeException("Insufficient stock for shipment.");
            }
            if (!configuration.getMainWarehouseCode().equals(shipmentData.getSourceLocation().getWarehouseCode())) {
                throw new RuntimeException("Только исходящие отгрузки из основного склада можно внешние.");
            }
            final LocationData locSrc = shipmentData.getSourceLocation();
            final LocationData locDst = shipmentData.getDestinationLocation();
            if (List.of(locDst.getCode(), locDst.getWarehouseCode(), locDst.getUnitOfMeasure()).stream().anyMatch(StringUtils::isBlank)) {
                throw new RuntimeException("Destination location should not be empty");
            }
            
            if (shipmentData.getWeight() != null && shipmentData.getWeight().compareTo(configuration.getMaxShipmentWeight()) > 0) {
                throw new RuntimeException("Shipment weight exceeds limit.");
            }
            
            Shipment externalShipment = shipmentRepos.save(buildExternalShipment(locSrc, locDst, shipmentData));
            ShipmentData externalShipmentData = ConvertersKt.toShipmentData(externalShipment, ShipmentData.ShipmentDirection.OUTBOUND, configuration.getMainWarehouseCode());
            kafkaProducer.sendOutgoingShipment(externalShipmentData);
            
            calculateDeliveryCost(externalShipmentData);
            
            return new ShipmentResponseDto("in_process", externalShipmentData);
        } catch (ValidationException e) {
            return new ShipmentResponseDto("FAILURE", shipmentData, e.getMessage());
        }
    }

    public void calculateDeliveryCost(ShipmentData shipmentData) {
        String url = "http://transport-logistics-service/api/v1/delivery/calculate";
        
        Map<String, Object> request = new HashMap<>();
        request.put("fromWarehouse", shipmentData.getSourceLocation().getWarehouseCode());
        request.put("toWarehouse", shipmentData.getDestinationLocation().getWarehouseCode());
        request.put("weight", shipmentData.getWeight());
        request.put("volume", shipmentData.getVolume());
        request.put("hazardous", shipmentData.getHazardousMaterial());
        
        try {
            ResponseEntity<Map> response = restTemplate.postForEntity(url, request, Map.class);
            
            if (response.getStatusCode().is2xxSuccessful()) {
                Map<String, Object> result = response.getBody();
                shipmentData.setDeliveryCost((BigDecimal) result.get("cost"));
                shipmentData.setDeliveryDays((Integer) result.get("estimatedDays"));
            }
        } catch (Exception e) {
            log.warn("Failed to calculate delivery cost: {}", e.getMessage());
        }
    }

    private Shipment buildExternalShipment(LocationData locSrc, LocationData locDst, ShipmentData shipmentData) {
        if (locDst == null) throw new ValidationException("Invalid destination location (not found)");
        if (locSrc == null) throw new ValidationException("Invalid source location (not found)");
        if (locDst.getUnitOfMeasure() == null) throw new NullPointerException("Invalid destination location (unit of measure not found)");
        if (locSrc.getUnitOfMeasure() == null) throw new NullPointerException("Invalid source location (unit of measure not found)");
        if (!locSrc.getUnitOfMeasure().equals(locDst.getUnitOfMeasure())) throw new ValidationException("Invalid shipment (different measurement units)");
        if (shipmentData.getHazardousMaterial() && !locDst.getHazardousStorageApproved()) {
            throw new ValidationException("Destination location not approved for hazardous materials");
        }
        // if (shipmentData.getQuantity().compareTo(BigDecimal.ZERO) <= 0) {
        //     throw new ValidationException("Shipment quantity must be positive");
        // }
        // if (locSrc.getCode().equals(locDst.getCode())) {
        //     throw new ValidationException("Source and destination cannot be the same");
        // }
        return Shipment.reserved(
                locSrc.getCode(),
                locDst.getCode(),
                DEFAULT_CONVERSION_FACTOR,
                  shipmentData.getQuantity(),
                  LocalDateTime.now(),
                  shipmentData.getDescription() != null
                        ? shipmentData.getDescription()
                        : "Shipment from external system"
        );
    }

    @Override
    public int processRejectedOutcommingShipmentsReester(Long shipmentId) {
        var foundShipment = shipmentRepos.findById(shipmentId);
        processLocationReservations(foundShipment.get().getSourceLocationCode());
        StorageLocation loc = locationRepository.getByLocationCode(foundShipment.getSourceLocationCode());
        Shipment[] reester = loc.getOutcommingShipments();
        int summa = 0;
        for(int k = 0; k < reester.length; k++) {
            if (!reester[k].getRejectionDate().isNull()) {
                if (reester[k].getShipmentType().getName() = "handling_fee") {
                    k+=1;
                    summa += calculateHandlingFee(foundShipment.getSourceLocationCode(), reester[k].getQuantity().intValue());
                } else {
                    BigDecimal a = reester[k].getQuantity();
                    summa += a.intValue();
                }
            }
        }
        return summa;
    }

    @Override
    public int processOutcommingShipmentsReester(Long shipmentId) {
        Shipment foundShipment = shipmentRepos.findById(shipmentId);
        StorageLocation loc = locationRepository.getByLocationCode(foundShipment.getSourceLocationCode());
        Shipment[] reester = loc.getOutcommingShipments();
        int summa = 0;
        for(int k = 1; k <= reester.length; k++) {
            if (reester[k].getRejectionDate().isNull()) {
                if (reester[k].getShipmentType().getName() = "europe") {
                BigDecimal a = reester[k].getQuantity();
                summa += a.intValue();
                }
            }
        }
        return summa;
    }

    @Override
    public void processOutcommingContragentWarehouseResponse(Long shipmentId, @NotNull ShipmentResponseDto shipmentResponse) {
        Optional<Shipment> foundShipment = shipmentRepos.findById(ConvertIdUtils.convertWarehouseCode(shipmentId));
        processLocationReservations(foundShipment.get().getSourceLocationCode());
        if ("ACCEPTED.OUTGOING".equals(shipmentResponse.getStatus())) {
            foundShipment.map(p -> {
                p.setAcceptanceDate(shipmentResponse.getData().getCreationDate());
                return p;
            }).ifPresent(shipmentRepos::save);
        } else {
            foundShipment.map(p -> {
                p.setRejectionDate(LocalDateTime.now());
                return p;
            }).ifPresent(shipmentRepos::save);
        }
    }

    @Override
    public void processOutcommingShipmentResponse(Long shipmentId, @NotNull ShipmentResponseDto shipmentResponse) {
        var foundShipment = shipmentRepos.findById(shipmentId);
        if ("ACCEPTED".equals(shipmentResponse.getStatus())) {
            foundShipment.map(p -> {
                p.setAcceptanceDate(shipmentResponse.getData().getCreationDate());
                return p;
            }).ifPresent(shipmentRepos::save);
        } else {
            foundShipment.map(p -> {
                p.setRejectionDate(LocalDateTime.now());
                return p;
            }).ifPresent(shipmentRepos::save);
        }
    }

    @Transactional
    private Optional<Restocking> processLocationReservations(String locCode) {
        try {
            locationRepository.findByCode(locCode)
                    .map(StorageLocation::getId)
                    .ifPresent(locId -> restockingService.processReservations(locId));
            return Optional.of(restockingService.calculateRestockingNeeds(locId));
        } catch (RestockingException ex) {
            log.warn(ex.getMessage());
        }
    }
}

Код примерный, для ознакомления, призван дать представление о количестве вариаций листингов, которые можно из него нарезать почти персонально под каждого.

На что можно было бы посмотреть в первую очередь:

• работа с DI;

• области видимости;

• AOP;

• магия в цифрах и буквах;

• общая стилистика и рисунок форм выражений, инструкций, методов;

• перегруженность смыслом, единственная ответственность, небольшая копипастность;

• проверки, fail-fast (так же тут), исключения.

Дальше, при более пристальном чтении не структуры кода, а каждого метода, я бы заметил:

• небезопасные вызовы методов у сомнительных данных, потенциальные места для NPE;

• интересные конвертации типов и сомнительного характера переменные;

• заподозрил бы необходимость транзакционности;

• поставил бы под сомнение уместность валидаций, не говоря уже о их имплементациях;

• может быть нашёл бы странное использование синтаксиса;

• повозмущался бы грубым игнорированием DRY, захотел бы что-то инкапсулировать наконец.

Более сложные ошибки выявляются после более глубокого изучения кода. Здесь на них намекать не стану.

package ru.logisticscompany.wms;

import java.math.BigDecimal;
import java.time.LocalDate;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

import static org.junit.jupiter.api.Assertions.assertEquals;

public class WarehouseStreamTest {

    record Product(
            String id, String name, String category, BigDecimal price, LocalDate expiryDate, boolean isHazardous
    ) {}

    record StorageLocation(
            String id,
            String name,
            String zoneType,
            BigDecimal capacity,
            BigDecimal currentQuantity,
            boolean allowsHazardous,
            List<Product> products
    ) {}

    record Warehouse(
            String id,
            String name,
            List<StorageLocation> locations
    ) {}

    private static final List<Warehouse> WAREHOUSES = List.of(
            new Warehouse(
                    "WH1", "Москва Центральный",
                    List.of(
                            new StorageLocation(
                                    "L1", "Зона приемки", "RECEIVING",
                                    new BigDecimal("1000"), new BigDecimal("150"),
                                    false,
                                    List.of(
                                            new Product("P1", "Ноутбук Lenovo", "Электроника", new BigDecimal("75000"), LocalDate.of(2025, 12, 31), false),
                                            new Product("P2", "Мышь Logitech", "Электроника", new BigDecimal("2500"), LocalDate.of(2026, 6, 30), false)
                                    )
                            ),
                            new StorageLocation(
                                    "L2", "Холодильная камера", "COLD",
                                    new BigDecimal("500"), new BigDecimal("300"),
                                    false,
                                    List.of(
                                            new Product("P3", "Молоко", "Молочные", new BigDecimal("85"), LocalDate.of(2024, 3, 15), false),
                                            new Product("P3", "Молоко", "Молочные", new BigDecimal("85"), LocalDate.of(2024, 3, 20), false)
                                    )
                            )
                    )
            ),
            new Warehouse(
                    "WH2", "Самара Логистик",
                    List.of(
                            new StorageLocation(
                                    "L3", "Стеллаж", "BULK",
                                    new BigDecimal("2000"), new BigDecimal("1800"),
                                    false,
                                    List.of(
                                            new Product("P4", "Кресло", "Мебель", new BigDecimal("15000"), null, false),
                                            new Product("P5", "Стол", "Мебель", new BigDecimal("9500"), null, false),
                                            new Product("P1", "Ноутбук Lenovo", "Электроника", new BigDecimal("75000"), LocalDate.of(2025, 12, 31), false)
                                    )
                            ),
                            new StorageLocation(
                                    "L4", "Хим.хранилище", "HAZARDOUS",
                                    new BigDecimal("300"), new BigDecimal("50"),
                                    true,
                                    List.of(
                                            new Product("P6", "Краска", "Химия", new BigDecimal("520"), LocalDate.of(2025, 8, 1), true),
                                            new Product("P7", "Растворитель", "Химия", new BigDecimal("320"), LocalDate.of(2024, 12, 31), true)
                                    )
                            )
                    )
            ),
            new Warehouse(
                    "WH3", "Новосибирск Северный",
                    List.of(
                            new StorageLocation(
                                    "L5", "Зона комплектации", "PICKING",
                                    new BigDecimal("800"), new BigDecimal("600"),
                                    false,
                                    List.of(
                                            new Product("P8", "Телевизор", "Электроника", new BigDecimal("55000"), LocalDate.of(2027, 5, 1), false),
                                            new Product("P9", "Наушники", "Электроника", new BigDecimal("12000"), LocalDate.of(2026, 10, 15), false),
                                            new Product("P2", "Мышь Logitech", "Электроника", new BigDecimal("2500"), LocalDate.of(2026, 6, 30), false)
                                    )
                            ),
                            new StorageLocation(
                                    "L6", "Огнеопасные", "HAZARDOUS",
                                    new BigDecimal("200"), new BigDecimal("30"),
                                    true,
                                    List.of(
                                            new Product("P10", "Лак для волос", "Косметика", new BigDecimal("450"), LocalDate.of(2025, 3, 1), true),
                                            new Product("P11", "Аэрозоль", "Химия", new BigDecimal("280"), LocalDate.of(2024, 11, 30), true)
                                    )
                            )
                    )
            )
    );

    @Test
    void testAllLevels() {
        LocalDate checkDate = LocalDate.of(2024, 3, 20);
         
       // Задание 1. Вывести список опасных товаров. 
        List<String> hazardousProducts = ...
        
        assertEquals(List.of("Аэрозоль", "Краска", "Лак для волос", "Растворитель"), hazardousProducts);
         
        // Задание 2. Склад с наибольшей общей стоимостью товаров 
        Optional<Map.Entry<String, BigDecimal>> richestWarehouse = ...
        
        assert richestWarehouse.isPresent();
        assertEquals("Москва Центральный", richestWarehouse.get().getKey());
        assertEquals(new BigDecimal("150170"), richestWarehouse.get().getValue());
        
        // Задание 3. Найти товары, которые есть на нескольких складах, с указанием складов     
        Map<String, List<String>> productsOnMultipleWarehouses =  ...
        
        assert productsOnMultipleWarehouses.containsKey("Ноутбук Lenovo");
        assertEquals(List.of("Москва Центральный", "Самара Логистик"), 
                productsOnMultipleWarehouses.get("Ноутбук Lenovo"));
        assert productsOnMultipleWarehouses.containsKey("Мышь Logitech");
        assertEquals(2, productsOnMultipleWarehouses.get("Мышь Logitech").size());
    }
}

Представленный код носит иллюстративный характер, но не отличается принципиально от реально заданных задач.

Первый алгоритм очень простой, околоджуновский уровень.

Главная задача – распрямить наборы вложенных коллекций в одну, отфильтровать по опасности каким-либо образом.

 List<String> hazardousProducts = WAREHOUSES.stream()
                .flatMap(warehouse -> warehouse.locations().stream()
                        .filter(loc -> "HAZARDOUS".equals(loc.zoneType()))
                        .flatMap(loc -> loc.products().stream()
                                .filter(Product::isHazardous)
                                .map(Product::name)))
                .distinct()
                .toList();

Тут возможно объединение условий с OR/AND, выделение в лямбды или приватные методы, наличие или отсутствие сортировки и уникальности.

Второй чуть сложнее, но кроме вложенности при обработке стримов не должен вызвать сложностей. Некоторые могут забыть как правильно суммировать, Интересный нюанс – суммирование композитного типа BigDecimal, не прямое суммирование элементов стрима.
Что важно:

• понимание, что из списка надо получать мапу или пару;

• выстроить по каждому складу набор стоимостей или цен (экономические термины не обсуждаем);

• рассчитанные значения поставить в соответствие с каким-то идентификатором склада;

• выстроить кастомную сортировку;

• взять один элемент, а не весь набор данных;

В отличие от первого варианта, тут выстраивается уже определенная последовательность действий, при нарушении которой сильно меняется результат.

Optional<Map.Entry<String, BigDecimal>> richestWarehouse = WAREHOUSES.stream()
                .collect(Collectors.toMap(
                        Warehouse::name,
                        warehouse -> warehouse.locations().stream()
                                .flatMap(loc -> loc.products().stream())
                                .map(Product::price)
                                .reduce(BigDecimal.ZERO, BigDecimal::add)
                ))
                .entrySet().stream()
                .max(Map.Entry.comparingByValue());

Посмотрим, что можно тут сделать.

Сначала надо сформировать алгоритм, текстом или псевдокодом. Можно насоздавать много переменных или приватных методов, для которых подстраивать вычисления. Важна структурированность мышления.

Вариантов тут может быть много. Например:

• Сначала можно за каждой записью товара закрепить идентификатор склада (имя, код, ID). Если подразумевается. что товар может быть на разных складах, то ведущее значение товар, вторичное – склад.

• Получается список пар значений (Map.Entry или Pair, кому как удобнее), нужно найти одинаковые ключи и сгруппировать, сформировав список из значений. Тут важно понимать, что есть терминальная операция группировки в мапу.

• Получили мапу идентификатором товаров на список складов. Надо посчитать количество складов для каждого товара и выбросить те, которые с 1 складом.

• Дальше отбросить склады и их количество, ведь нас интересует только товар.

Map<String, List<String>> productsOnMultipleWarehouses = WAREHOUSES
                .parallelStream()
                .flatMap(warehouse -> warehouse.locations().stream()
                        .flatMap(loc -> loc.products().stream()
                                .map(product -> new AbstractMap.SimpleEntry<>(
                                        product.name(),
                                        warehouse.name()
                                ))
                        )
                )
                .collect(Collectors.groupingByConcurrent(
                        Map.Entry::getKey,
                        ConcurrentHashMap::new,
                        Collectors.mapping(
                                Map.Entry::getValue,
                                Collectors.toList()
                        )
                ))
                .entrySet().stream()
                .filter(entry -> {
                    long uniqueWarehouseCount = entry.getValue().stream().distinct().count();
                    return uniqueWarehouseCount > 1;
                })
                .collect(Collectors.toMap(
                        Map.Entry::getKey,
                        entry -> entry.getValue().stream()
                                .distinct()
                                .toList()))
                ));

Дальше можно обсуждать параллелизм, его уместность, потокобезопасность и другие интересные, но малополезные здесь инструменты и явления.

Логично выделить несколько этапов при решении:

• Понять, что один набор данных прогоняется от начала до конце один раз, а второй надо переначитывать заново время от времени. То есть с одним набором надо работать как с обычным итератором, а для второго сохранять итератор мало, надо сохранить ещё и контейнер с элементами.

• Вспомнить или логически подойти к тому, что нужны методы next() и hasNext() в итераторе. 

• В качестве первого приближения предположить например, что итератор работает с другими итераторами, сформировать конструктор и вызов конструктора.

• Сформировать тело метода hasNext(), тут достаточно простая логика.

• Начать писать тело метода next(). Тут рано или поздно профессионал поймёт, что обратить вспять данные, выдаваемые итератором, невозможно. Значит, надо хранить ещё состояние Iteable.

• Добавить поле с состоянием, изменить конструктор и его вызов.

• Дальше возникнет вопрос как сбрасывать счетчик позиции у Iterable, будут вопросы по простому копированию данных (возможно несколько вариантов).

• И финальная часть – понять как инициализировать и изменять те переменные, которые хранят состояния наборов данных values и transform.

Вот как, например, я могу реализовать этот алгоритм, если не хочу тратить слишком много времени для подготовки статьи:

private class IteratorFunc<T>(
        private val iteratorTransform: Iterator<(T) -> T>,
        private val iterableValue: Iterable<T>
    ): Iterator<T> {
        private var currentTransform: ((T)->T) = if (iteratorTransform.hasNext()) { iteratorTransform.next() } else { it: T -> it }
        private var iteratorValue = iterableValue.toList().iterator()

        override fun hasNext() = iteratorValue.hasNext() || iteratorTransform.hasNext()

        override fun next(): T = if (iteratorValue.hasNext()) { currentTransform(iteratorValue.next()) } else {
            currentTransform = iteratorTransform.next()
            iteratorValue = iterableValue.toList().iterator()
            currentTransform(iteratorValue.next())
        }

    }

Не предполагается, что сразу будет предложено идеальное решение, итеративный подход вполне себе жизнеспособен.