Когда через 2 недели я нашел решение для моей проблемы, о которой говорилось выше, я решил, что запишу ее для других и для себя в будущем.
Я не смог воспроизвести проблему, потому что aws command-
aws s3 cp --recursive dir s3: // input-queue / dir
почему-то не загружало сообщения достаточно быстро, чтобы это могло вызвать стресс в моей локальной линии данных , Итак, я отключил конвейер данных, и как только в очереди появилось 10 тыс. Сообщений SQS, я запустил его и, как и ожидалось, вылетел из-за ошибки Out Of Memory после обработки ~ 3000 сообщений. Оказывается, что конвейер был в состоянии обрабатывать непрерывную пропускную способность, но он сломался, когда он начал с загрузкой сообщений 10 000.
Моя гипотеза состояла в том, что проблема происходит, потому что Java сборщик мусора не может должным образом очистить объекты после исполнения. Итак, я начал анализировать сгенерированный дамп кучи и после нескольких дней исследований наткнулся на возможную root причину ошибки Out of Memory. Было около 5000 экземпляров моего класса MessageHandlerTask, когда в идеале они должны были быть G C 'd после обработки и не продолжать накапливаться.
Дальнейшее изучение этого направления привело меня к root причина - оказалось, что код использует Executors.newFixedThreadPool () для создания ExecutorService для отправки задач. В этой реализации использовалась неограниченная очередь задач, поэтому, если было отправлено слишком много задач, все они ждали в очереди, занимая огромную память. Реальность была похожа - сообщения опрашивались быстрее, чем они могли быть обработаны. Это привело к тому, что было создано много допустимых экземпляров MessageHandlerTask, которые заполнили кучу памяти, если имелось невыполнение сообщения.
Исправление было связано с созданием ThreadPoolExecutor с ArrayBlockingQueue емкостью 100, так что на количество экземпляров MessageHandlerTask и его переменных-членов.
Разобравшись с исправлением, я перешел к оптимизации конвейера для максимальной пропускной способности, изменяя MaximumPoolSize ThreadPoolExecutor. Оказалось, что некоторые исключения SQS-соединения происходят при большем количестве потоков. Дальнейшее расследование показало, что увеличение размера пула соединений SQS уменьшило эту проблему. В итоге я остановился на количестве 40 потоков для заданного размера кучи Xmx 1,5 ГБ и 80 пулов соединений SQS, чтобы потоки задач не исчерпали соединения SQS во время обработки. Это помогло мне достичь пропускной способности 44 сообщений / с только с одним экземпляром конвейера данных.
Я также выяснил, почему пакетная линия никогда не ломалась в Production, несмотря на то, что страдала от аналогичной реализации ExecutorService - оказывается, что конвейер данных мог Подчеркните, что слишком много одновременных загрузок S3, но сообщения для batchpipeline были созданы datapipeline постепенно. Кроме того, пакетная линия имела намного более высокую пропускную способность, которую я оценил на уровне 347 сообщений / с при использовании 70 MaximumPoolSize.