Merge pull request #102 from zendesk/ensure_binlog_row_image

don't allow maxwell to start when binlog_row_image == minimal

src/main/java/com/zendesk/maxwell/MaxwellMysqlStatus.java

@@ -15,41 +15,46 @@ private String sqlStatement(String variableName) {
 		return "SHOW VARIABLES LIKE '" + variableName + "'";
 	}
 
-	private String getVariableState(String variableName) throws SQLException, MaxwellCompatibilityError {
+	private String getVariableState(String variableName, boolean throwOnMissing) throws SQLException, MaxwellCompatibilityError {
 		ResultSet rs;
 
 		rs = connection.createStatement().executeQuery(sqlStatement(variableName));
 		String status;
 		if(!rs.next()) {
-			throw new MaxwellCompatibilityError("Could not check state for Mysql variable: "+variableName);
+			if ( throwOnMissing ) {
+				throw new MaxwellCompatibilityError("Could not check state for Mysql variable: " + variableName);
+			} else {
+				return null;
+			}
 		}
 
 		status = rs.getString("Value");
 		return status;
 	}
 
-	private void ensureNotReadOnly() throws SQLException, MaxwellCompatibilityError {
-		if (!getVariableState("read_only").equals("OFF")) {
-			throw new MaxwellCompatibilityError("read_only must be OFF.");
+	private void ensureVariableState(String variable, String state) throws SQLException, MaxwellCompatibilityError
+	{
+		if (!getVariableState(variable, true).equals(state)) {
+			throw new MaxwellCompatibilityError("variable " + variable + " must be set to '" + state + "'");
 		}
 	}
 
-	private void ensureReplicationOn() throws SQLException, MaxwellCompatibilityError {
-		if (!getVariableState("log_bin").equals("ON")) {
-			throw new MaxwellCompatibilityError("log_bin status must be ON.");
-		}
-	}
 
-	private void ensureReplicationFormatROW() throws SQLException, MaxwellCompatibilityError {
-		if (!getVariableState("binlog_format").equals("ROW")) {
-			throw new MaxwellCompatibilityError("binlog_format must be ROW.");
-		}
+	private void ensureRowImageFormat() throws SQLException, MaxwellCompatibilityError {
+		String rowImageFormat = getVariableState("binlog_row_image", false);
+		if ( rowImageFormat == null ) // only present in mysql 5.6+
+			return;
+
+		if ( !rowImageFormat.equals("FULL"))
+			throw new MaxwellCompatibilityError("binlog_row_image must be FULL.");
 	}
 
 	public static void ensureMysqlState(Connection c) throws SQLException, MaxwellCompatibilityError {
 		MaxwellMysqlStatus m = new MaxwellMysqlStatus(c);
-		m.ensureNotReadOnly();
-		m.ensureReplicationOn();
-		m.ensureReplicationFormatROW();
+
+		m.ensureVariableState("read_only", "OFF");
+		m.ensureVariableState("log_bin", "ON");
+		m.ensureVariableState("binlog_format", "ROW");
+		m.ensureRowImageFormat();
 	}
 }