skip JSON over 16kb

Author: hzarka

src/main/java/com/zendesk/maxwell/schema/columndef/ColumnDef.java

@@ -22,7 +22,7 @@
 public abstract class ColumnDef implements Cloneable {
 	protected static final Interner INTERNER = Interners.newWeakInterner();
 	private static final DynamicEnum dynamicEnum = new DynamicEnum(Byte.MAX_VALUE);
-	private String name;
+	protected String name;
 	private final byte type;
 	private short pos;
 

src/main/java/com/zendesk/maxwell/schema/columndef/JsonColumnDef.java

@@ -3,10 +3,14 @@
 import com.github.shyiko.mysql.binlog.event.deserialization.json.JsonBinary;
 import com.zendesk.maxwell.producer.MaxwellOutputConfig;
 import com.zendesk.maxwell.row.RawJSONString;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 
 public class JsonColumnDef extends ColumnDef {
+	private static final Logger LOGGER = LoggerFactory.getLogger(JsonColumnDef.class);
+
 	private JsonColumnDef(String name, String type, short pos) {
 		super(name, type, pos);
 	}
@@ -25,7 +29,12 @@ public Object asJSON(Object value, MaxwellOutputConfig config) throws ColumnDefC
 		} else if ( value instanceof byte[] ){
 			try {
 				byte[] bytes = (byte[]) value;
-				jsonString = bytes.length > 0 ? JsonBinary.parseAsString(bytes) : "null";
+				if (bytes.length > 16384) {
+					LOGGER.warn("Skipping JSON over limit for field: " + name);
+					jsonString = "null";
+				} else {
+					jsonString = bytes.length > 0 ? JsonBinary.parseAsString(bytes) : "null";
+				}
 				return new RawJSONString(jsonString);
 			} catch (IOException e) {
 				throw new RuntimeException(e);