Kafka-connect-jdbc: JDBC Source : Schema owner cannot be specified - task fails with error java.lang.IllegalArgumentException: Number of groups must be positive.

[rmoff]

Scenario: User wants to read data from a schema that it doesn’t own. In this example, for conciseness, we’re using SYSTEM as the user’s credentials, and testmgr as the target schema from which we want to read data.

Confluent Platform 3.2

Steps to reproduce:

Make Oracle JDBC driver available to Connect:

cp ojdbc8.jar confluent-3.2.0/share/java/kafka-connect-jdbc/

Set up Oracle with test user and dummy table with a row of data

sql system/oracle@localhost:1521/orcl

create user testmgr identified by oracle;
grant create session to testmgr;
grant resource to testmgr;
alter user testmgr quota unlimited on users;

create table testmgr.accounts (c1 int);
insert into testmgr.accounts values (0);
commit;

Launch Connect: (assumes ZK, Kafka, Schema Registry already running)

./bin/connect-distributed etc/kafka/connect-distributed.properties

Create connector that works - Connecting as the table owner (testmgr), and accessing the table with no prefix.

Validate config:

PUT /connector-plugins/JdbcSourceConnector/config/validate HTTP/1.1
Content-Type: application/json
Host: localhost:8083
Connection: close
User-Agent: Paw/3.0.16 (Macintosh; OS X/10.12.4) GCDHTTPRequest
Content-Length: 227

{"name":"oracle-accounts-testmgr","connector.class":"JdbcSourceConnector","connection.url":"jdbc:oracle:thin:testmgr/[email protected]:1521/orcl","mode":"bulk","table.whitelist":"accounts","topic.prefix":"oracle-accounts-"}

0 errors. Create connector:

PUT /connectors/oracle-accounts-testmgr/config HTTP/1.1
Content-Type: application/json
Host: localhost:8083
Connection: close
User-Agent: Paw/3.0.16 (Macintosh; OS X/10.12.4) GCDHTTPRequest
Content-Length: 227

{"name":"oracle-accounts-testmgr","connector.class":"JdbcSourceConnector","connection.url":"jdbc:oracle:thin:testmgr/[email protected]:1521/orcl","mode":"bulk","table.whitelist":"accounts","topic.prefix":"oracle-accounts-"}

Runs fine, creates topic & row of data as expected

Now validate connector which connects as one user (SYSTEM) and tries to read data from another:

PUT /connector-plugins/JdbcSourceConnector/config/validate HTTP/1.1
Content-Type: application/json
Host: localhost:8083
Connection: close
User-Agent: Paw/3.0.16 (Macintosh; OS X/10.12.4) GCDHTTPRequest
Content-Length: 233

{"name":"oracle-accounts-system","connector.class":"JdbcSourceConnector","connection.url":"jdbc:oracle:thin:system/[email protected]:1521/orcl","mode":"bulk","table.whitelist":"testmgr.accounts","topic.prefix":"oracle-accounts-"}

0 Errors. Create connector:

PUT /connectors/oracle-accounts-system/config HTTP/1.1
Content-Type: application/json
Host: localhost:8083
Connection: close
User-Agent: Paw/3.0.16 (Macintosh; OS X/10.12.4) GCDHTTPRequest
Content-Length: 233

{"name":"oracle-accounts-system","connector.class":"JdbcSourceConnector","connection.url":"jdbc:oracle:thin:system/[email protected]:1521/orcl","mode":"bulk","table.whitelist":"testmgr.accounts","topic.prefix":"oracle-accounts-"}

Connect task fails:

[2017-04-20 21:44:03,628] ERROR Failed to reconfigure connector's tasks, retrying after backoff: (org.apache.kafka.connect.runtime.distributed.DistributedHerder:926)
java.lang.IllegalArgumentException: Number of groups must be positive.
        at org.apache.kafka.connect.util.ConnectorUtils.groupPartitions(ConnectorUtils.java:42)
        at io.confluent.connect.jdbc.JdbcSourceConnector.taskConfigs(JdbcSourceConnector.java:127)
        at org.apache.kafka.connect.runtime.Worker.connectorTaskConfigs(Worker.java:230)
        at org.apache.kafka.connect.runtime.distributed.DistributedHerder.reconfigureConnector(DistributedHerder.java:967)
        at org.apache.kafka.connect.runtime.distributed.DistributedHerder.reconfigureConnectorTasksWithRetry(DistributedHerder.java:918)
        at org.apache.kafka.connect.runtime.distributed.DistributedHerder.access$800(DistributedHerder.java:101)
        at org.apache.kafka.connect.runtime.distributed.DistributedHerder$17$1.call(DistributedHerder.java:931)
        at org.apache.kafka.connect.runtime.distributed.DistributedHerder$17$1.call(DistributedHerder.java:928)
        at org.apache.kafka.connect.runtime.distributed.DistributedHerder.tick(DistributedHerder.java:250)
        at org.apache.kafka.connect.runtime.distributed.DistributedHerder.run(DistributedHerder.java:200)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
        at java.lang.Thread.run(Thread.java:745)

[rmoff]

Workaround : use query instead of table.whitelist. Example config:

{
  "name": "oracle-accounts-query",
  "connector.class": "JdbcSourceConnector",
  "connection.url": "jdbc:oracle:thin:system/[email protected]:1521/orcl",
  "mode": "bulk",
  "query": "select * from testmgr.accounts",
  "topic.prefix": "oracle-accounts-query-"
}

[skitching-um]

I have also encountered this problem at a large customer. Enabling DEBUG level logging shows the problem clearly:

• 2017-06-30 10:51:52,430 DEBUG io.confluent.connect.jdbc.source.TableMonitorThread: Got the following tables: [SOMETABLE1, SOMETABLE2]
• 2017-06-30 10:51:52,695 DEBUG io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier: TimestampIncrementingTableQuerier{name='SOMETABLE1', query='null', topicPrefix='xyz', timestampColumn='LASTMOD', incrementingColumn='USER_ID'} prepared SQL query: SELECT * FROM "SOMETABLE1" WHERE ....

However SOMETABLE1 is in schema MYSCHEMA which is not the default for the current user, so the select fails...

[spencerarogers]

We're also having this problem with Oracle schemas using the 3.3.0 JDBC source connector. We can load tables just fine from the schema of which our user is an owner, but other schemas from which the user has permissions to read cannot be queried. We have tried all permutations of specifying the "schema.pattern" config and qualifying the table whitelist with the schema name. Per the workaround above, we can get the query option to work, i.e. "query": "SELECT * FROM FOO.BAR".

When we specify "table.whitelist"="FOO.BAR" the config shows up properly in the logs:

INFO JdbcSourceConnectorConfig values:
...
table.whitelist = [FOO.BAR]

but in our exception we're seeing

ERROR Failed to run query for table TimestampIncrementingTableQuerier{name='BAR'
, query='null', ...}: {} (io.confluent.connect.jdbc.source.JdbcSourceTask)
java.sql.SQLSyntaxErrorException: ORA-00942: table or view does not exist

It seems like somehow the schema qualifier is getting stripped in between initializing the connector with it's config and generating the query from the table whitelist. I've been searching for the root cause in the source code but have thus far been unsuccessful.

It seems like being forced to use the query configuration forces us to have 1 connector instance per table, and we're working with hundreds of tables with limited hardware and I'm worried we won't be able to scale this solution.

Also, we eventually get the same java.lang.IllegalArgumentException: Number of groups must be positive. as above after the java.sql.SQLSyntaxErrorException: ORA-00942: table or view does not exist occurs.

[skitching-um]

Some JDBC drivers support a URL parameter which specifies the default schema to use. This can be used to work around the problem.

[TattiQ]

@spencer1248 , what did you decide to use in the end? I stumbled upon the same issue and I agree 1 connector instance per table is an overkill.

[spencerarogers]

@TattiQ One of my coworkers threw a hack into the JDBC connector but unfortunately I don't have access to them or the source code at the moment.

[vinaykasam]

I am facing similar issue any help is appreciated

[aspektr]

If you try extract data from mssql view you''ll get
ERROR Failed to reconfigure connector's tasks, retrying after backoff: (org.apache.kafka.connect.runtime.distributed.DistributedHerder) java.lang.IllegalArgumentException: Number of groups must be positive.

You have to use query like "query": "select * from report.dbo.blacklisted" instead of "table.whitelist"