java.lang.OutOfMemoryError

[bchazalet]

I running kafka-connect jdbc on one table, for the first time, and I am quickly running out of memory even though I have set:

batch.max.rows = 1
tasks.max = 1

It looks like the batch limit is not being effective in my case (I am running it against a postgres database).

Does the batch limit apply the very first time kafka-connect runs?

[ewencp]

batch.max.rows controls how much memory is used to buffer data by the connector itself during a call to its poll() method. What it does is generate and run a query, but if more than batch.max.rows are generated, it will stop processing after that many and return from poll(). This guarantees that if you have a large number of rows, it doesn't try to process them all at once.

However, this doesn't do anything like adding a LIMIT clause to the query. Perhaps we need an option to control that as well (or reuse batch.max.rows, although whether that can be done safely depends on the query mode you're using). However, I wouldn't have expected this to cause an out of memory error as the results should just stream in from the DB.

What database and JDBC driver are you using, and could you share anything about your connector configuration/use case that might shed any more light on how it is hitting the OOM issue?

[bchazalet]

I am using postgresq 9.4 and the driver is the one from the confluent 2.0.0 package, i.e postgresql-9.4-1206-jdbc41.jar.

About the use case, I am using the jdbc connector against an existing table in the database, for the first time. The table has got around 10 millions rows, even though I rather care about the future updates rather than the data already in it.

I am using the timestamp mode.

Leaving the sensitive information out, here's my properties file:

name=test
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=10
batch.max.rows=100
mode=timestamp
timestamp.column.name=modified_timestamp
table.whitelist=mytable
connection.url=jdbc:postgresql://**********

[bchazalet]

I think the problems lies with the streaming.

We have internal projects where we are streaming the data (from postgres as well) and we had to have autoCommit set to false on the connection, as well as setting the setFetchDirection to ResultSet.FETCH_FORWARD and setFetchSize to some value on the prepared statement:

I don't know how portable that is across databases though, and if you're already doing something similar or not.

[SleepyBrett]

Having the exact same issue against postgres (PostgreSQL) 9.5.0 using the driver in the confluent 2.0 platform.

My config looks like this currently:
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
"connection.url": "jdbc:postgresql://localhost:5432/state",
"tasks.max": "1",
"name": "jdbctest-source",
"topic.prefix": "connect-test",
"mode": "timestamp+incrementing",
"incrementing.column.name": "auto_id",
"timestamp.column.name": "modified",
"table.whitelist": "cpos_xxx",
"batch.max.rows": "10"
}

[kadrach]

Same issue here. Postgres 9.3 with the current confluent platform.

[SleepyBrett]

I, not knowing much about java, jdbc or the jdbc postgres driver have done a few hacks as suggested by bchazalet that seems to have solved the issue with postgres.

You can check and hack on my branch at: https://github.com/SleepyBrett/kafka-connect-jdbc/tree/postgres-fixes

Fair warning, I haven't tested this with any other driver.

[Astn]

We were having a similar issue with MySQL and found some JDBC connection string properties we could use to alter the way the driver was functioning.
Maybe something similar would work for the Postgres issue?

Here is an example config that solved our Out Of Memory issue with MySQL.

{"name": "test-mysql-jdbc-source",
                            "config": {
                              "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
                              "tasks.max": "10",
                              "topic.prefix": "test-import-",
                              "connection.url":"jdbc:mysql://mysql-test-server:3306/testdatabase?user=...&password=...&defaultFetchSize=10000&useCursorFetch=true",
                              "table.whitelist":"bigtable1,bigtable2,bigtable3",
                              "poll.interval.ms":"86400000",
                              "mode":"bulk"}}

Notice that we added defaultFetchSize=10000&useCursorFetch=true to the JDBC connection string.

Hopefully this helps someone.

[bchazalet]

@Astn I didn't know you could put all that stuff in the connection string, that's a very good tip!

[Astn]

@bchazalet Just took a quick peek at the postgres JDBC connection string parameters here:

https://jdbc.postgresql.org/documentation/head/connect.html

And I noticed this, which looks like it might help you.

defaultRowFetchSize = int

Determine the number of rows fetched in ResultSet by one fetch with trip to the database. Limiting the number of rows are fetch with each trip to the database allow avoids unnecessary memory consumption and as a consequence OutOfMemoryException.

The default is zero, meaning that in ResultSet will be fetch all rows at once. Negative number is not available.

Either way, I'm curious to know what ends up working for you.

Cheers

[jasonjho]

@Astn @bchazalet Also curious about defaultRowFetchSize. I'm assuming you still have to set autoCommit=false on the Statement in code?

[Astn]

@jasonjho I haven't tried that. But I would first exhaust all the connection string properties options.
The docs aren't very good, but adding readOnly=true seems like something else that may help.

readOnly = boolean

Put the connection in read-only mode

I think we will need to put some of this on the wiki anyhow.

[timyguo]

I have run into the same issue while bulk loading tables into from postgres, I've included
readOnly=true&defaultRowFetchSize=10000
into my connection string and the issue still persists.

[davidchen322]

I think that one problem is we need to enable streaming ResultSets when the results are too large. It appears to be a little different on each database, although it looks like in all DBs, we need to set the fetch size (stmt.setFetchSize) so the JDBC driver doesn't try to load the ENTIRE query result into memory.

We're using MySQL, and we've found that by modifying all the TableQuerier implementation (BulkTableQuerier,java and TimestampIncrementingTableQuerier.java) PreparedStatement create's to do the following prompts MySQL's JDBC Driver to do a "Streaming" ResultSet. There are caveats with this approach too, but this will work. See the MySql JDBC Driver Documentation, in the ResultSet section, for more details

db.prepareStatement(queryString,
  java.sql.ResultSet.TYPE_FORWARD_ONLY,
  java.sql.ResultSet.CONCUR_READ_ONLY);
...
...
stmt.setFetchSize(Integer.MIN_VALUE);
...

We have not yet tried the solution here (#34 (comment)), but it is a quick workaround with zero code changes necessary (which is nice)

[gauravnanda79]

I had the same issue while working with MS SQL Server. I had to set autocommit to false for this to get working. By default autocommit is enabled in which case the fetchDirection and fetchSize are ignored.

Since the driver is anyways intended to be readonly the explicit setAutocommit(false) should go into the codebase

[utahkay]

Are there any plans on fixing this for Postgres, or has someone found a workaround? The solution in #34 worked for me with MySQL, but I have tried autoCommit=false and defaultRowFetchSize=1000 on postgres and still run out of memory.