fix oom. if config resendRequestChunkSize. Loading data is batch

quickfixj-core/src/main/java/quickfix/Session.java

@@ -2333,8 +2333,13 @@ private void nextLogon(Message logon) throws FieldNotFound, RejectLogon, Incorre
     }
 
     private void resendMessages(Message receivedMessage, int beginSeqNo, int endSeqNo)
-            throws IOException, InvalidMessage, FieldNotFound {
-
+            throws IOException, FieldNotFound {
+        // Prevent a one-time load of data that is too large, resulting in memory OOM。 if config resendRequestChunkSize。
+        int lastEndSeqNoSent = resendRequestChunkSize == 0 ? endSeqNo : beginSeqNo + resendRequestChunkSize - 1;
+        if (lastEndSeqNoSent > endSeqNo) {
+            lastEndSeqNoSent = endSeqNo;
+        }
+        endSeqNo = lastEndSeqNoSent;
         final ArrayList<String> messages = new ArrayList<>();
         try {
             state.get(beginSeqNo, endSeqNo, messages);

quickfixj-core/src/test/java/quickfix/SessionFactoryTestSupport.java

@@ -44,6 +44,19 @@ public static Session createSession(SessionID sessionID, Application application
                 .build();
     }
 
+    public static Session createSession(SessionID sessionID, Application application,
+                                        boolean isInitiator, boolean resetOnLogon,
+                                        boolean validateSequenceNumbers,
+                                        int resendRequestChunkSize, boolean enableNextExpectedMsgSeqNum) {
+        return new Builder().setSessionId(sessionID).setApplication(application).setIsInitiator(isInitiator)
+                .setResetOnLogon(resetOnLogon).setValidateSequenceNumbers(validateSequenceNumbers)
+                .setPersistMessages(true)
+                .setResendRequestChunkSize(resendRequestChunkSize)
+                .setEnableNextExpectedMsgSeqNum(enableNextExpectedMsgSeqNum)
+                .build();
+    }
+
+
     public static Session createSession(SessionID sessionID, Application application,
                                         boolean isInitiator, boolean resetOnLogon, boolean validateSequenceNumbers,
                                         boolean useDataDictionary, DefaultApplVerID senderDefaultApplVerID) {
@@ -108,7 +121,7 @@ public static final class Builder {
         private final boolean forceResendWhenCorruptedStore = false;
         private final Set<InetAddress> allowedRemoteAddresses = null;
         private final boolean validateIncomingMessage = true;
-        private final int resendRequestChunkSize = 0;
+        private int resendRequestChunkSize = 0;
         private boolean enableNextExpectedMsgSeqNum = false;
         private final boolean enableLastMsgSeqNumProcessed = false;
         private final boolean validateChecksum = true;
@@ -239,5 +252,10 @@ public Builder setEnableNextExpectedMsgSeqNum(final boolean enableNextExpectedMs
             this.enableNextExpectedMsgSeqNum = enableNextExpectedMsgSeqNum;
             return this;
         }
+
+        public Builder setResendRequestChunkSize(final int resendRequestChunkSize) {
+            this.resendRequestChunkSize = resendRequestChunkSize;
+            return this;
+        }
     }
 }

quickfixj-core/src/test/java/quickfix/SessionTest.java

@@ -3145,4 +3145,47 @@ public void testSend_ShouldKeepPossDupFlagAndOrigSendingTime_GivenAllowPosDupCon
         assertTrue(sentMessage.getHeader().isSetField(PossDupFlag.FIELD));
         assertTrue(sentMessage.getHeader().isSetField(OrigSendingTime.FIELD));
     }
+
+
+    @Test
+    public void testSessionWithResendRequestChunkSizeAndEnableNextExpectedMsgSeqNum() throws Exception {
+        final UnitTestApplication application = new UnitTestApplication();
+
+        final SessionID sessionID = new SessionID(
+                FixVersions.BEGINSTRING_FIX44, "SENDER", "TARGET");
+        try (Session session = createSession(sessionID, application, true,
+                true, false,10,true)) {
+
+            final UnitTestResponder responder = new UnitTestResponder();
+            session.setResponder(responder);
+
+            session.logon();
+            session.next();
+
+            final Message logonRequest = new Message(responder.sentMessageData);
+            Message logonResponse = createLogonResponse(sessionID, logonRequest, 1);
+            // config Field NextExpectedMsgSeqNum=1
+            logonResponse.setField(new NextExpectedMsgSeqNum(1));
+            session.next(logonResponse);
+
+            assertEquals(
+                    1,
+                    application.lastToAdminMessage().getHeader()
+                            .getInt(MsgSeqNum.FIELD));
+            assertEquals(2, session.getStore().getNextTargetMsgSeqNum());
+            assertEquals(2, session.getStore().getNextSenderMsgSeqNum());
+
+            session.next(createHeartbeatMessage(1002));
+            assertNotEquals(ResendRequest.MSGTYPE, application
+                    .lastToAdminMessage().getHeader().getString(MsgType.FIELD));
+
+            session.next(createHeartbeatMessage(1003));
+            assertNotEquals(ResendRequest.MSGTYPE, application
+                    .lastToAdminMessage().getHeader().getString(MsgType.FIELD));
+
+            session.next(createHeartbeatMessage(1001));
+            assertNotEquals(ResendRequest.MSGTYPE, application
+                    .lastToAdminMessage().getHeader().getString(MsgType.FIELD));
+        }
+    }
 }