Kafka-connect-jdbc: support postgresql boolean/BIT

[bchazalet]

I know you are already aware of this, as it's mentioned in the code but it'd be nice to handle the BIT type for postgresql.

Indeed, boolean columns get seen as BIT types and fail while being converted:

[2016-01-08 16:09:48,292] WARN Ignoring record due to SQL error: (io.confluent.connect.jdbc.JdbcSourceTask:73)
org.postgresql.util.PSQLException: Bad value for type byte : f
    at org.postgresql.jdbc2.AbstractJdbc2ResultSet.getByte(AbstractJdbc2ResultSet.java:2100)
    at io.confluent.connect.jdbc.DataConverter.convertFieldValue(DataConverter.java:281)
    at io.confluent.connect.jdbc.DataConverter.convertRecord(DataConverter.java:68)
    at io.confluent.connect.jdbc.TimestampIncrementingTableQuerier.extractRecord(TimestampIncrementingTableQuerier.java:165)
    at io.confluent.connect.jdbc.JdbcSourceTask.poll(JdbcSourceTask.java:211)
name=control-info-pgstaging
    at org.apache.kafka.connect.runtime.WorkerSourceTask$WorkerSourceTaskThread.execute(WorkerSourceTask.java:353)
    at org.apache.kafka.connect.util.ShutdownableThread.run(ShutdownableThread.java:82)

The resulting avro message ends up with all boolean values being null (and also I think these failures might slow down the import process quite a lot).

[ewencp]

@bchazalet Yeah, I think different DBs handle these differently. Longer term I think we may end up with specialized code for different JDBC drivers in order to handle everything well.

We've got some system tests that help us make sure we get these conversions right. The current test covers MySQL, but that's just the one we wrote first. I'd love to get more DBs tested (especially the couple of most popular ones).

[bchazalet]

@ewencp thanks for your quick reply.

It's particularly annoying for postgresql, as boolean types are commonly used and will systematically fail down that route.

[bchazalet]

Actually, one might argue that a sql BIT matches an Avro boolean better than a Avro int8. That's what I'm investigating in that branch.

[lepfhty]

+1 for treating BIT as boolean

[rhauch]

BIT of length 1 should be treated as a boolean, but most DBMSes support other lengths and those much more closely map (at least in Java) to java.util.BitSet. JDBC even makes it fairly easy to distinguish with java.sql.Types.BIT and the length of the column.

Unfortunately Kafka Connect Schemas have no corresponding built-in type with similar semantics, though that'd be easy to add since BitSet has methods to convert to and from byte[] containing a little-endian representation of all the bits in this bit set.

[ewencp]

@rhauch A new logical type seems like the right solution here if it can just be encoded as a byte[]. There's a definite tradeoff between too many basic types (lots of overhead of writing serializers and any type of conversions) and robust, complete coverage of logical types. In this case, I think the width is fixed, right? So a logical schema for a BitSet that uses byte[] as the underlying type and has a parameter for width seems like the right solution.

That could definitely be added to Kafka Connect itself if we think it's broadly useful. I'd like to validate that it would show up in a variety of connectors before getting it into the core system though, since even logical types have some overhead for authors of converters (and for the framework itself!).

[rhauch]

@ewencp What's the best way to proceed with that discussion, and perhaps to include other potential logical types? Open a JIRA issue for Kafka Connect? Or start a discussion on the (developer?) mailing list first?

[ewencp]

@rhauch JIRA for Connect. We can discuss on the mailing list, although until now it's basically just been committer discretion re: which types are worthwhile. If we have a reasonable proposal w/ a couple of (theoretical) use cases w/ other systems, I'm happy to help you get it into Kafka Connect.

[rhauch]

I posted to the Kafka dev mailing list with a proposal for supporting custom logical types.

[pushingice]

Any movement on this? We want to use Confluent with our real, live, production Postgres instance. This is an actual feature we want. EDIT: We have many boolean columns across many tables.

[gfodor]

another +1 on this one, I don't think that a generalized BitSet/logical type system is the solution here if I understand the problem correctly -- the root cause here sounds like PostgreSQL's JDBC driver reporting boolean columns as length-1 BIT columns. Regardless of any future support for arbitrary BitSets (something which in my experience is a rare column type in schemas anyhow,) don't we want PostgreSQL boolean columns to be treated as booleans downstream, not as length-1 BitSets? It seems what is needed is to simply interpret length-1 BIT columns as booleans (as is done in bchazalets's patch) potentially only doing this when the underlying database driver is PostgreSQL. Doing this would unblock using PostgreSQL with Kafka Connect, which seems pretty important :)

[jasonjho]

Agree with @gfodor here on treating length-1 BIT columns as booleans. We have a major production system running on PostgreSQL and would love for this to be unblocked as we're anxious to get our hands on Connect. Any way to accelerate this would be super helpful. Thanks!

[pushingice]

I have a version working here for boolean and jsonb types: Postgres Fork

It's probably too much of a hack, but I would submit a PR if anyone from confluent wants it.

[iceNuts]

Hi there, I am wondering how is this ticket going? I have the same issue when using with Postgres. I think it is still not fixed for now.

[jasonjho]

Second this, @iceNuts. We've forked and patched our own fixes internally based on some of the strategies mentioned above, but it would be nice for this to be addressed in the next release!

[JerryAlbinA]

Hi All,

we are facing same issue !! , is there any solution ?

my log

[2017-02-02 08:22:14,810] WARN Ignoring record due to SQL error: (io.confluent.connect.jdbc.source.JdbcSourceTask:78)
org.postgresql.util.PSQLException: Bad value for type byte : t
at org.postgresql.jdbc2.AbstractJdbc2ResultSet.getByte(AbstractJdbc2ResultSet.java:2100)
at io.confluent.connect.jdbc.source.DataConverter.convertFieldValue(DataConverter.java:286)
at io.confluent.connect.jdbc.source.DataConverter.convertRecord(DataConverter.java:73)
at io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier.extractRecord(TimestampIncrementingTableQuerier.java:184)
at io.confluent.connect.jdbc.source.JdbcSourceTask.poll(JdbcSourceTask.java:195)
at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:155)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:140)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:175)
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)

[bala2e]

@JerryAlbinA me too getting the same issue .
could you figure out the solution.
Please help me to fix it

[ryananguiano]

@bala2e @JerryAlbinA I got it to work with #261 by copying the fix from @pushingice onto updated code.

Here's how I loaded it into a docker image for production: https://github.com/revpoint/confluent-docker-images/tree/master/kafka-connect

[fsimond]

This issue should be easier to fix when DBMS-specific dialects will be added #303

[rhauch]

I just added a fix for this to #333. See the PostgreSQL dialect changes for details. Hope to merge that in the next few days.