From 2ad6714442c19b18db1906b30b770b2a9158ec4c Mon Sep 17 00:00:00 2001 From: Leonard Xu Date: Mon, 7 Mar 2022 12:00:43 +0800 Subject: [PATCH] [mysql] Fix ArrayIndexOutOfBoundsException in mysql binlog read phase (#911) --- .../binlog/io/BufferedSocketInputStream.java | 80 +++++++++++++++++++ .../io/BufferedSocketInputStreamTest.java | 53 ++++++++++++ 2 files changed, 133 insertions(+) create mode 100644 flink-connector-mysql-cdc/src/main/java/com/github/shyiko/mysql/binlog/io/BufferedSocketInputStream.java create mode 100644 flink-connector-mysql-cdc/src/test/java/com/github/shyiko/mysql/binlog/io/BufferedSocketInputStreamTest.java diff --git a/flink-connector-mysql-cdc/src/main/java/com/github/shyiko/mysql/binlog/io/BufferedSocketInputStream.java b/flink-connector-mysql-cdc/src/main/java/com/github/shyiko/mysql/binlog/io/BufferedSocketInputStream.java new file mode 100644 index 000000000..d11fa0e55 --- /dev/null +++ b/flink-connector-mysql-cdc/src/main/java/com/github/shyiko/mysql/binlog/io/BufferedSocketInputStream.java @@ -0,0 +1,80 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.github.shyiko.mysql.binlog.io; + +import java.io.FilterInputStream; +import java.io.IOException; +import java.io.InputStream; + +/** + * Copied from https://github.com/osheroff/mysql-binlog-connector-java project to fix + * https://github.com/ververica/flink-cdc-connectors/issues/460. + * + *

Line 50 and Line 70 ~ 72: Returns -1 means reach the end of InputStream. We should remove this + * class after we bumped a higher mysql-binlog-connector-java version where the + * https://github.com/osheroff/mysql-binlog-connector-java/issues/66 has been fixed. + */ +public class BufferedSocketInputStream extends FilterInputStream { + + private byte[] buffer; + private int offset; + private int limit; + + public BufferedSocketInputStream(InputStream in) { + this(in, 512 * 1024); + } + + public BufferedSocketInputStream(InputStream in, int bufferSize) { + super(in); + this.buffer = new byte[bufferSize]; + } + + @Override + public int available() throws IOException { + return limit == -1 ? in.available() : limit - offset + in.available(); + } + + @Override + public int read() throws IOException { + if (offset < limit) { + return buffer[offset++] & 0xff; + } + offset = 0; + limit = in.read(buffer, 0, buffer.length); + return limit != -1 ? buffer[offset++] & 0xff : -1; + } + + @Override + public int read(byte[] b, int off, int len) throws IOException { + if (offset >= limit) { + if (len >= buffer.length) { + return in.read(b, off, len); + } + offset = 0; + limit = in.read(buffer, 0, buffer.length); + if (limit == -1) { + return limit; + } + } + int bytesRemainingInBuffer = Math.min(len, limit - offset); + System.arraycopy(buffer, offset, b, off, bytesRemainingInBuffer); + offset += bytesRemainingInBuffer; + return bytesRemainingInBuffer; + } +} diff --git a/flink-connector-mysql-cdc/src/test/java/com/github/shyiko/mysql/binlog/io/BufferedSocketInputStreamTest.java b/flink-connector-mysql-cdc/src/test/java/com/github/shyiko/mysql/binlog/io/BufferedSocketInputStreamTest.java new file mode 100644 index 000000000..50c8ff2d4 --- /dev/null +++ b/flink-connector-mysql-cdc/src/test/java/com/github/shyiko/mysql/binlog/io/BufferedSocketInputStreamTest.java @@ -0,0 +1,53 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.github.shyiko.mysql.binlog.io; + +import org.junit.Test; + +import java.util.Arrays; + +import static org.junit.Assert.assertEquals; + +/** Unit test for {@link BufferedSocketInputStream}. */ +public class BufferedSocketInputStreamTest { + + @Test + public void testReadFromBufferedSocketInputStream() throws Exception { + BufferedSocketInputStream in = + new BufferedSocketInputStream( + new ByteArrayInputStream( + new byte[] {'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H'})); + byte[] buf = new byte[3]; + assertEquals(3, in.read(buf, 0, buf.length)); + Arrays.equals(new byte[] {'A', 'B', 'C'}, buf); + assertEquals(5, in.available()); + + assertEquals(3, in.read(buf, 0, buf.length)); + Arrays.equals(new byte[] {'D', 'E', 'F'}, buf); + assertEquals(2, in.available()); + + assertEquals(2, in.read(buf, 0, buf.length)); + Arrays.equals(new byte[] {'G', 'H'}, buf); + assertEquals(0, in.available()); + + // reach the end of stream normally + assertEquals(-1, in.read(buf, 0, buf.length)); + assertEquals(0, in.available()); + } +}