DDL parser: Allow stored procedure variables in LIMIT clause

org.apache.kafka.connect.errors.ConnectException: Multiple parsing errors
io.debezium.text.ParsingException:
no viable alternative at input 'CREATE DEFINER=`devadmin`@`%` PROCEDURE `auto_closure_detail_new`(IN I_COMPANY_ID INT,IN I_TERMINATION_TYPE varchar(1),IN I_USER_ID varchar(10),IN I_LOAN_ID INT(10),IN LIMIT1 INT(10),IN LIMIT2 INT(10),OUT O_ERRORFLAG varchar(1),OUT O_ERRORMSG varchar(1000))
BEGIN
DECLARE V_LOAN_ID		INT(20);
DECLARE V_BUSINESS_DATE		DATE;
DECLARE O_BALANCE_PRINCIPAL DECIMAL(22,4); 	
DECLARE O_OVERDUE_PRINCIPAL DECIMAL(22,4); 
DECLARE O_OVERDUE_INSTALLMENTS DECIMAL (22,4);      
DECLARE O_UNBILLED_INSTALLMENTS DECIMAL(22,4); 	
DECLARE O_OTHER_DUES DECIMAL (22,4);
DECLARE O_GAP_INTEREST DECIMAL (22,4); 	
DECLARE O_PREPAY_PENALTY DECIMAL (22,4);
DECLARE O_LOCKIN_PERIOD_INTEREST DECIMAL (22,4);
DECLARE O_OD_AMOUNT DECIMAL (22,4);
DECLARE O_OTHER_PAYABLES DECIMAL(22,4); 	 
DECLARE O_SECURE_DEPOSIT DECIMAL(22,4); 
DECLARE O_SECURE_DEPOSIT_INTEREST DECIMAL(22,4);
DECLARE O_SECURE_DEPOSIT_TDS DECIMAL(22,4);
DECLARE O_SECURE_DEPOSIT_GAP_INTEREST DECIMAL(22,4); 
DECLARE O_SECURE_DEPOSIT_GAP_TDS DECIMAL(22,4);
DECLARE O_ADVANCE_EMI_REFUNDS DECIMAL(22,4);
DECLARE O_TERMINATION_ID INT(10) ;
DECLARE V_NET_REC_PAY DECIMAL(22,4);
DECLARE V_NET_REC_PAY1 DECIMAL(22,4);
DECLARE V_LOAN_CUSTOMER_ID INT(10) ;
DECLARE V_SCHEME_ID INT(10) ;
DECLARE V_MANUAL_ADVICE_ID INT(10);
DECLARE O_TXNADVICE_ID INT(10);
DECLARE V_AUTO_CLOSE_AMOUNT_LIMIT DECIMAL(22,4);
DECLARE V_LOAN_TERM_COUNT INT(10);
DECLARE V_MANUAL_COUNT INT(10);
DECLARE PROC_PARAM VARCHAR(200);
DECLARE DONE   INT DEFAULT 0;
DECLARE V_PAY_AMOUNT DECIMAL(22,4);
DECLARE V_LOAN_NO VARCHAR(50);\r
DECLARE V_CHARGE_AMOUNT DECIMAL(22,4);
DECLARE V_INCOME_BOOKING  DECIMAL(22,4);
DECLARE O_REJECTRECORD INT(10);
DECLARE O_UPLOADEDRECORD INT(10);
DECLARE CLOSURE_LOAN_CUR CURSOR FOR SELECT A.LOAN_ID, A.LOAN_CUSTOMER_ID,a.loan_scheme,A.LOAN_NO
				FROM CR_LOAN_DTL A, CR_SCHEME_M B,GCD_CUSTOMER_M G  
				WHERE B.SCHEME_ID=A.LOAN_SCHEME AND G.CUSTOMER_ID=A.LOAN_CUSTOMER_ID  
				AND B.TERMINATION_ALLOWED='Y' AND A.REC_STATUS='A' AND ((A.LOAN_REPAYMENT_TYPE='I' AND A.DISBURSAL_STATUS='F') OR (A.LOAN_REPAYMENT_TYPE<>'I' AND A.DISBURSAL_STATUS='P')) 
				AND A.LOAN_ID NOT IN (SELECT LOAN_ID FROM CR_TERMINATION_DTL WHERE REC_STATUS IN ('P','F','A') ) AND ((A.LOAN_REPAYMENT_TYPE='I' 
				AND A.LOAN_MATURITY_DATE < V_BUSINESS_DATE))
				and LOAN_ID = IF(I_LOAN_ID=0, LOAN_ID, I_LOAN_ID)
				 AND a.loan_scheme in (select scheme_id from cr_closure_master_dtl where rec_status='A') 
				-- and a.LOAN_MATURITY_DATE='2020-03-07'
			 -- AND a.loan_scheme in (289,290)
				order by A.LOAN_ID desc limit LIMIT1'

io.debezium.text.ParsingException: mismatched input 'select' expecting <EOF>
at io.debezium.connector.mysql.MySqlConnectorTask.start(MySqlConnectorTask.java:298)                                                                         
at io.debezium.connector.common.BaseSourceTask.start(BaseSourceTask.java:106)          
at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:213)  
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:184)     
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:234)        
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)

io.debezium.text.ParsingException: mismatched input 'select' expecting <EOF>
   at io.debezium.antlr.AntlrDdlParser.throwParsingException(AntlrDdlParser.java:351)
   at io.debezium.antlr.AntlrDdlParser.parse(AntlrDdlParser.java:93)
   at io.debezium.relational.history.AbstractDatabaseHistory.lambda$recover$1(AbstractDatabaseHistory.java:115)
   at io.debezium.relational.history.KafkaDatabaseHistory.recoverRecords(KafkaDatabaseHistory.java:307)
   at io.debezium.relational.history.AbstractDatabaseHistory.recover(AbstractDatabaseHistory.java:82)
   at io.debezium.connector.mysql.MySqlSchema.loadHistory(MySqlSchema.java:268)
   at io.debezium.connector.mysql.MySqlTaskContext.loadHistory(MySqlTaskContext.java:164)
   at io.debezium.connector.mysql.MySqlConnectorTask.start(MySqlConnectorTask.java:110)

database.history.skip.unparseable.ddl: true
event.processing​.failure.handling.mode: skip
inconsistent.schema.handling.mode: skip