Kafka-connect-jdbc: 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 (https://github.com/confluentinc/kafka-connect-jdbc/issues/34#issuecomment-203042361), 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.

[iceNuts]

I am having the same issue on Postgres even if adding readOnly=true&autoCommit=false&defaultFetchSize=10&useCursorFetch=true

[rach]

This bug make kafka connect unusable with Postgres. Is there an solution which worked except the fork from @SleepyBrett ??

[HarvinderBhullar]

I am getting the issue with Postgres.. I tried all the suggestions mentioned above

[seanlindo]

Is there any progress on this?

[larskluge]

Apparently this is not an issue about the number of rows in the database, but how much data is in each row, meaning the number of columns and data in each cell selected.

@seanlindo The patch discussed above works and @astubbs built a version for me that worked for me. Feel free to use it while this ticket is still open:

kafka-connect-jdbc-lars-3.2.0.jar.zip

[seanlindo]

Thank you @larskluge! I actually ended up using the latest version from Blue Apron which includes a ton of other quality of life fixes, like supporting the Postgres boolean type.

https://github.com/blueapron/kafka-connect-jdbc

[seanlindo]

@larskluge On an unrelated-but-maybe-related note, how does connect know that this jdbc plugin is different from the one that comes installed? They have identical classnames.

[larskluge]

@seanlindo This jar is almost identical to the original and should replace the original jar, therefore identical classnames.

Thanks for pointing to Blue Apron's, that might work better anyway.

Also, I'm about to try http://debezium.io — perhaps another good alternative.

[seanlindo]

@larskluge Ah, it never dawned on me to simply replace the jar completely. Debezium is absolutely on my radar as well.

[rdp]

For followers, be careful, for Postgres, the connection string must be &defaultRowFetchSize=xx (notice the word Row in there). And yes autocommit must (annoyingly) be turned off for it to work in Postgres, as well.

[fsimond]

I have the same problem with a postgres database.

We should be able to disable autocommit, or Kafka Connect should do it itself if the fetch size is defined, as it is done in Spark SQL:
https://github.com/apache/spark/blob/0ae96495dedb54b3b6bae0bd55560820c5ca29a2/sql/core/src/main/scala/org/apache/spark/sql/jdbc/PostgresDialect.scala#L97

Otherwise, Kafka Connect is unusable with Postgres. This should be easier to fix with #303

[vgw-rhysc]

Any movement on this? Using kafka-connect-jdbc on an existing Postgres database seems to be somewhat flawed?

[kkitts]

I am hitting this issue as well - in a Netezza database (which is a fork of Postres).

[rhauch]

Also, there will be improvements after the 4.1 release. Specifically, we're adding DBMS-specific dialects (#333) that will make it easier to customize the JDBC behavior for certain DBMSes.

[kkitts]

Hi Randall,

I am familiar with the Netezza JDBC parameters. I tried combinations of setting auto commit, readonly and batchSize but I could not get anything to work.

However, the alter user with rowsetlimit does work exactly as advertised - so that is at least a nice temporary workaround. I should be able to now use the timestamp and/or incrementing modes in conjunction with this rowsetlimit and a SQL statement, right?

Thanks again for your help - very much appreciated.

Best Regards,

Kevin Kitts

On Mar 14, 2018, at 7:17 PM, Randall Hauch [e-mail] wrote:

Also, there will be improvements after the 4.1 release. Specifically, we're adding DBMS-specific dialects (#333 https://github.com/confluentinc/kafka-connect-jdbc/pull/333) that will make it easier to customize the JDBC behavior for certain DBMSes.

—
You are receiving this because you commented.
Reply to this email directly, view it on GitHub https://github.com/confluentinc/kafka-connect-jdbc/issues/34#issuecomment-373206578, or mute the thread https://github.com/notifications/unsubscribe-auth/ASsVWgeRfc2BtUJ363H1ZrRH4LobskfAks5teaUIgaJpZM4HBNlw.

[Deninc]

Also, there will be improvements after the 4.1 release. Specifically, we're adding DBMS-specific dialects (#333) that will make it easier to customize the JDBC behavior for certain DBMSes.

@rhauch do you have any timeline when the 4.1 release is?

[kkitts]

Hello All,

I wanted to follow up on this thread. Based on a comment I found on Stackoverflow, I managed to set the database I am working with (Netezza) to limit the number of rows returned in a query which avoided the Java out of memory error.

The SQL Command I used was:

ALTER USER with ROWSETLIMIT 200

However, compared to another non-streaming tool that I am using (Pentaho) the performance of loading into Elasticsearch is quite poor. With Pentaho I can specify much larger row counts to get in each query of the database without any out of memory errors.

In order to optimize the throughput we need much more control to set very large buffers on the JDBC source query. Is there any workaround that would allow us to vastly increase the buffer size so that much larger batches could be supported?

[andybryant]

@rhauch - I can confirm using a Dialect with Confluent 5.0 does fix the issue.
See https://gist.github.com/andybryant/8b6b532963a723bb1cfb5ddf8c9dd53f
Put that class in a jar with the property file at META-INF/services so that the connector can find it, put the jar in the kafka-connect-jdbc directory

If you ramp up the logging io.confluent.connect.jdbc=DEBUG you can confirm whether connect using your dialect over the standard Postgres one...

connect_1          | [2018-08-03 11:46:55,072] DEBUG Using dialect BatchingPostgreSqlDatabaseDialect with score 200 against JDBC subprotocol 'postgresql' and source 'jdbc:postgresql://mydb.ecorp:5432/pkafka?currentSchema=old_accounts&defaultRowFetchSize=1000&readOnly=true' (io.confluent.connect.jdbc.dialect.DatabaseDialects)

[rhauch]

@andybryant, thanks for the feedback. Would it always be better for PostgreSQL dialect to disable auto commit? If so, would you want to submit a PR with that change?

[andybryant]

Good question. JdbcDbWriter always turns auto commit off for the sink so it wouldn't affect that.
The question is whether turning off auto commit has any adverse effects in general for the Postgres source. I'm not a PostgreSQL expert so couldn't say for sure but I'm not aware of any.

Happy to submit a PR to update the standard Postgres dialect.

[hagmonk]

@andybryant @rhauch since the source side of the connector is literally intended to be read only and not making any DML statements, there's nothing to "commit", making it a no-op to disable autocommit.

However I don't think it's sufficient to just disable auto commit. Per the blueapron fork and the pgjdbc documentation, the statement fetch size must also be set. This can also be set as a default in the JDBC connection string, but I think making it an explicit connector property with an explicitly sensible value (rather than 0) would make everyone's life easier:

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.

[kkitts]

In order to maximize throughput, it would be better to make fewer calls to the database with larger values of the RowFetchSize than a large number of calls with a small RowFetchSize.

Would be very helpful if it were possible to increase the buffer sizes so that large row sets could be returned in a single call.

[vinod-bb]

defaultFetchSize=10000&useCursorFetch=true

Did this work for timestamp mode ?

[robertmackenzie]

I ran into this issue retrieving large result sets from SQL Server. Result sets are supposed to be streamed from SQL Server using adaptive buffering. In the end, I discovered the check for a valid connection causes the whole result set to be read into memory.

I figured that's because, without support for multiple active result sets on a connection, you need to read your entire current result set into memory before you can execute a new one to check if the connection is valid. Sure enough, after some painful time with the debugger, I wound up up in this code, which "detaches" the result set to run a new command.

Replacing connection.isValid() with connection.isClosed() gets around the problem. You have to watch out though; since a connection can be invalid and not properly closed, you can be left hanging indefinitely if there's a connection problem. I got around that with the SQL Server specific timeout socketTimeout, which will throw and cause a querier reset and reconnect.

I haven't checked this is the issue with other databases, or replicated this on SQL Server with server cursors rather than adaptive buffering, but thought I'd give a shout.

[somasundaramsekar]

For Postgres use the latest driver, connection string properties defaultRowFetchSize are not available with the older drivers

Properties reference: https://github.com/pgjdbc/pgjdbc

[briancharous]

Are there any updates or workarounds on this in the past 6 months? I'm running into the same issue with Postgres and defaultRowFetchSize doesn't solve it.

[ivanyu]

We have solved this issue in this fork. It requires only setting one parameter in the database connection URL (documentation).
Hope this helps someone!

[ashikaumanga]

We also ran into this issue and we fixed this by adding a new config parameter to Kafka Connect Jdbc Source connector.
Inside the connector we set the fetchSize of the preparedSTatement.

stmt.setFetchSize(resultSetFetchSize);

This worked fine for MySql server with MariaDB JDBC Driver.
Do you think this is can be a generic solutions since most of the JDBC Drivers respect fetchSize parameter?

We did the fix for Cp 3.3.x
https://github.com/ashikaumanga/kafka-connect-jdbc/pull/1/files

[psarka]

The magic for MS SQL Server that worked for me was ;selectMethod=cursor, as in

connection.url=jdbc:sqlserver://10.1.1.1:1433;databaseName=database;selectMethod=cursor

This issue is very disturbing.

[yohei1126]

The PR is merged so this issue is fixed on master branch? https://github.com/confluentinc/kafka-connect-jdbc/pull/758

[gharris1727]

The fix in #758 was also ported to other databases and tested with an embedded MySQL instance in #775.
Closing as fixed. Will be released in 5.2.3, 5.3.3, and 5.4.1 when they are released.