使用StreamSets从Kafka Consumer向MySQL写入数据遇阻求助

Hey there! Let's dig into why you're having trouble writing Kafka consumer data to MySQL via JDBC—since you can already read from databases and write to text files, your core consumer logic is likely working fine. The issue is almost certainly tied to something specific to the JDBC write workflow. Here's a step-by-step breakdown to troubleshoot:

Since you mentioned using a Kafka Consumer, I'll cover both custom code scenarios and Kafka Connect (if you're using the JDBC Sink Connector):

If You're Using Custom Consumer + JDBC Code #

• Double-check you're executing the write statement
  It's easy to create a PreparedStatement and set parameters, then forget to call executeUpdate()—this is a super common oversight. Example of the fix:

  String insertSql = "INSERT INTO your_table (field1, field2) VALUES (?, ?)";
  try (Connection conn = DriverManager.getConnection(dbUrl, dbUser, dbPass);
       PreparedStatement pstmt = conn.prepareStatement(insertSql)) {
      pstmt.setString(1, kafkaMessage.getField1());
      pstmt.setInt(2, kafkaMessage.getField2());
      // Don't forget this line!
      int rowsAffected = pstmt.executeUpdate();
      System.out.println("Wrote " + rowsAffected + " row(s) to MySQL");
  } catch (SQLException e) {
      // Print detailed errors here—don't swallow exceptions!
      e.printStackTrace();
  }
  

• Validate transaction handling
  If you've disabled auto-commit with conn.setAutoCommit(false), you must call conn.commit() after executing your write. Without this, the transaction will roll back silently.
• Verify database user permissions
  Just because your user can read from MySQL doesn't mean they have write access. Run this query in MySQL to check:

  SHOW GRANTS FOR 'your_db_user'@'your_host';
  

  Make sure you see INSERT, UPDATE, or ALL PRIVILEGES for your target database/table.

If You're Using Kafka Connect JDBC Sink #

• Audit your connector configuration
  Key settings to verify:
  • connector.class=io.confluent.connect.jdbc.JdbcSinkConnector (or the correct sink class for your setup)
  • connection.url, connection.user, connection.password are correct (same as your working read config, but confirm write permissions)
  • auto.create=true or auto.evolve=true (if you need the connector to create/update tables; if not, ensure the target table exists with matching schema)
  • insert_mode: Default is insert, which will fail on duplicate primary keys. Use upsert (requires pk.fields config) or replace if duplicates are expected.
• Check Connect worker logs
  Kafka Connect logs are gold for troubleshooting—look in your Connect installation's logs/ directory (usually connect.log). You'll find specific errors like missing tables, schema mismatches, or permission issues here.

A lot of write failures happen without obvious errors because exceptions are swallowed or logs are missing.

• In custom code, never catch SQLException without printing the full stack trace or detailed error messages (including e.getMessage() and e.getSQLState()).
• For Kafka Connect, set the log level to DEBUG temporarily if you need more granular details about what the connector is doing.

Isolate the problem by writing a tiny standalone JDBC program to insert a test row into MySQL (no Kafka involved). Example:

public class JdbcWriteTest {
    public static void main(String[] args) throws SQLException {
        String dbUrl = "jdbc:mysql://localhost:3306/your_db?useSSL=false&serverTimezone=UTC";
        String dbUser = "your_user";
        String dbPass = "your_pass";
        
        try (Connection conn = DriverManager.getConnection(dbUrl, dbUser, dbPass);
             Statement stmt = conn.createStatement()) {
            int rows = stmt.executeUpdate("INSERT INTO test_table (test_field) VALUES ('Hello from JDBC')");
            System.out.println("Inserted " + rows + " row(s)");
        }
    }
}

• If this fails: The issue is with your JDBC setup, MySQL configuration, or user permissions—fix this first before bringing Kafka back into the mix.
• If this succeeds: The problem is in how your Kafka consumer is passing data to JDBC (e.g., message parsing errors, thread-safety issues with JDBC connections).

MySQL logs can reveal issues that JDBC doesn't surface clearly:

• Error log: Look for entries about denied connections, table corruption, or invalid queries. It's usually located at /var/log/mysql/error.log (Linux) or in your MySQL data directory.
• General query log: Enable it temporarily to see if MySQL is even receiving your write requests:

  SET GLOBAL general_log = 'ON';
  

  Check the log file (use SHOW VARIABLES LIKE 'general_log_file'; to find its path) to see if your INSERT/UPDATE statements are arriving. If they're not, the problem is in your JDBC code/Connect config. If they are but have errors, you'll see the exact MySQL error there.

内容的提问来源于stack exchange，提问作者Anurag Gupta