问:
MySql query fails in Kafka-connect
I am using Kafka connect (confluentinc/cp-kafka-connect:5.4.0) and have MySQL connector installed in it. Basically the following Dockerfile:

```<code>FROM confluentinc/cp-kafka-connect:5.4.0

RUN echo "===> Installing MySQL connector" \
    && curl -k -SL "https://dev.mysql.com/get/Downloads/Connector-J/mysql-connector-java-5.1.39.tar.gz" | tar -xzf - -C /usr/share/java/kafka-connect-jdbc/ --strip-components=1 mysql-connector-java-5.1.39/mysql-connector-java-5.1.39-bin.jar
</code>```

I created a connector with the following configurations:

curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
        "name": "test-connector",
        "config": {
                "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
                "connection.url":"jdbc:mysql://10.xx.yy.z:3306/demo_db?user=user&password=pass",
                "topic.prefix": "test-connector-",
                "mode":"incrementing",
                "query":"$QUERY",
                "incrementing.column.name": "eventId",
                "validate.non.null": false
                }
        }'
```

<p>I tried replacing the "$QUERY" mentioned above with the following query (pasting a formatted version for better readability):</p>

<pre class="lang-sql prettyprint-override"><code>SELECT * FROM (
    select DISTINCT t1.id,
            t1.name,
            t1.email,
            t1.department,
            t1.modified
        from
            test as t1
            LEFT OUTER JOIN test as t2 ON t1.id > -1
        WHERE
            t1.id > -1) something
</code>```

The above connector gets created and works as expected. However, the below query fails. and

SELECT * FROM(
        SELECT
            HSC.id eventId,
            HSC.actions,
            HSC.cause,
            HSC.metaData,
            CRED.uid uidCard,
            CRED.is_primary isPrimary,
            CRED.type cardType,
            PR.keys_ prin,
            HOUSE.uid orgUid,
            CHILD.uid childUid,
            USER.first_name fName,
            USER.last_name lName,
            CHILD.biz_phone bizPh,
            CHILD.cell_phone cellPh,
            AOF.state aof_state,
            RECORD.uid replacedCardUid
        FROM
            house_of_cards as HSC
            INNER JOIN play_card as CRED ON HSC.debit_card_id = CRED.id
            INNER JOIN prin PR ON CRED.prin_id = PR.id
            INNER JOIN child CHILD ON CRED.employee_id = CHILD.id
            INNER JOIN user USER ON USER.id = CHILD.user_id
            INNER JOIN house HOUSE ON CHILD.house_id = HOUSE.id
            LEFT JOIN address SHPADDR ON CRED.shipping_address = SHPADDR.id
            LEFT JOIN address AOF ON CRED.address_on_file = AOF.id
            LEFT JOIN play_card RECORD ON CRED.replaced_card_id = RECORD.id
        WHERE
        HSC.type = 'cardActionKamen' ) playcards
```

<p><em>EDIT: 1. Posting full error log</em>  (with modified sensitive data of course)</p>

```
[2020-02-16 20:31:46,463] INFO AbstractConfig values:
        batch.max.rows = 100
        catalog.pattern = null
        connection.attempts = 3
        connection.backoff.ms = 10000
        connection.password = null
        connection.url = jdbc:mysql://10.xx.yy.z:3306/demo_db?user=user&password=pass
        connection.user = null
        db.timezone = UTC
        dialect.name =
        incrementing.column.name = eventId
        mode = incrementing
        numeric.mapping = null
        numeric.precision.mapping = false
        poll.interval.ms = 5000
        query = SELECT * FROM( SELECT HSC.id eventId, HSC.actions, HSC.cause, HSC.metaData, CRED.uid uidCard, CRED.is_primary isPrimary, CRED.type cardType, PR.keys_ prin, HOUSE.uid orgUid, CHILD.uid childUid, USER.first_name fName, USER.last_name lName, CHILD.biz_phone bizPh, CHILD.cell_phone cellPh, AOF.state aof_state, RECORD.uid replacedCardUid FROM house_of_cards as HSC INNER JOIN play_card as CRED ON HSC.debit_card_id = CRED.id INNER JOIN prin PR ON CRED.prin_id = PR.id INNER JOIN child CHILD ON CRED.employee_id = CHILD.id INNER JOIN user USER ON USER.id = CHILD.user_id INNER JOIN house HOUSE ON CHILD.house_id = HOUSE.id LEFT JOIN address SHPADDR ON CRED.shipping_address = SHPADDR.id LEFT JOIN address AOF ON CRED.address_on_file = AOF.id LEFT JOIN play_card RECORD ON CRED.replaced_card_id = RECORD.id WHERE HSC.type = 'cardActionKamen' ) playcards
        quote.sql.identifiers = ALWAYS
        schema.pattern = null
        table.blacklist = []
        table.poll.interval.ms = 60000
        table.types = [TABLE]
        table.whitelist = []
        timestamp.column.name = []
        timestamp.delay.interval.ms = 0
        topic.prefix = card-view-test
        validate.non.null = false
 (org.apache.kafka.common.config.AbstractConfig)
[2020-02-16 20:31:46,595] INFO AbstractConfig values:
 (org.apache.kafka.common.config.AbstractConfig)
[2020-02-16 20:31:46,601] INFO [Worker clientId=connect-1, groupId=alpha-core-kafka-connect-wrapper] Connector card-view-test config updated (org.apache.kafka.connect.runtime.distributed.DistributedHerder)
[2020-02-16 20:31:46,602] INFO [Worker clientId=connect-1, groupId=alpha-core-kafka-connect-wrapper] Rebalance started (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator)
[2020-02-16 20:31:46,602] INFO [Worker clientId=connect-1, groupId=alpha-core-kafka-connect-wrapper] (Re-)joining group (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
[2020-02-16 20:31:46,609] INFO [Worker clientId=connect-1, groupId=alpha-core-kafka-connect-wrapper] Successfully joined group with generation 295 (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
[2020-02-16 20:31:46,609] INFO [Worker clientId=connect-1, groupId=alpha-core-kafka-connect-wrapper] Joined group at generation 295 with protocol version 2 and got assignment: Assignment{error=0, leader='connect-1-d82cc98e-5
619-4972-ad78-3b0a53c3b5bb', leaderUrl='http://10.xx.yy.36:8083/', offset=433, connectorIds=[card-view-test], taskIds=[card-view-test-0], revokedConnectorIds=[], revokedTaskIds=[], delay=0} with rebalance delay: 0 (org.apache
.kafka.connect.runtime.distributed.DistributedHerder)
[2020-02-16 20:31:46,609] INFO [Worker clientId=connect-1, groupId=alpha-core-kafka-connect-wrapper] Starting connectors and tasks using config offset 433 (org.apache.kafka.connect.runtime.distributed.DistributedHerder)
[2020-02-16 20:31:46,609] INFO [Worker clientId=connect-1, groupId=alpha-core-kafka-connect-wrapper] Starting connector card-view-test (org.apache.kafka.connect.runtime.distributed.DistributedHerder)
[2020-02-16 20:31:46,609] INFO [Worker clientId=connect-1, groupId=alpha-core-kafka-connect-wrapper] Starting task card-view-test-0 (org.apache.kafka.connect.runtime.distributed.DistributedHerder)
[2020-02-16 20:31:46,610] INFO Creating task card-view-test-0 (org.apache.kafka.connect.runtime.Worker)
[2020-02-16 20:31:46,610] INFO ConnectorConfig values:
        config.action.reload = restart
        connector.class = io.confluent.connect.jdbc.JdbcSourceConnector
        errors.log.enable = false
        errors.log.include.messages = false
        errors.retry.delay.max.ms = 60000
        errors.retry.timeout = 0
        errors.tolerance = none
        header.converter = null
        key.converter = null
        name = card-view-test
        tasks.max = 1
        transforms = []
        value.converter = null
 (org.apache.kafka.connect.runtime.ConnectorConfig)
[2020-02-16 20:31:46,610] INFO EnrichedConnectorConfig values:
        config.action.reload = restart
        connector.class = io.confluent.connect.jdbc.JdbcSourceConnector
        errors.log.enable = false
        errors.log.include.messages = false
        errors.retry.delay.max.ms = 60000
        errors.retry.timeout = 0
        errors.tolerance = none
        header.converter = null
        key.converter = null
        name = card-view-test
        tasks.max = 1
        transforms = []
        value.converter = null
 (org.apache.kafka.connect.runtime.ConnectorConfig$EnrichedConnectorConfig)
[2020-02-16 20:31:46,610] INFO Creating connector card-view-test of type io.confluent.connect.jdbc.JdbcSourceConnector (org.apache.kafka.connect.runtime.Worker)
[2020-02-16 20:31:46,610] INFO ConnectorConfig values:
        config.action.reload = restart
        connector.class = io.confluent.connect.jdbc.JdbcSourceConnector
        errors.log.enable = false
        errors.log.include.messages = false
        errors.retry.delay.max.ms = 60000
        errors.retry.timeout = 0
        errors.tolerance = none
        header.converter = null
        key.converter = null
        name = card-view-test
        tasks.max = 1
        transforms = []
        value.converter = null
 (org.apache.kafka.connect.runtime.ConnectorConfig)
[2020-02-16 20:31:46,610] INFO EnrichedConnectorConfig values:
        config.action.reload = restart
        connector.class = io.confluent.connect.jdbc.JdbcSourceConnector
        errors.log.enable = false
        errors.log.include.messages = false
        errors.retry.delay.max.ms = 60000
        errors.retry.timeout = 0
        errors.tolerance = none
        header.converter = null
        key.converter = null
        name = card-view-test
        tasks.max = 1
        transforms = []
        value.converter = null
 (org.apache.kafka.connect.runtime.ConnectorConfig)
[2020-02-16 20:31:46,610] INFO EnrichedConnectorConfig values:
        config.action.reload = restart
        connector.class = io.confluent.connect.jdbc.JdbcSourceConnector
        errors.log.enable = false
        errors.log.include.messages = false
        errors.retry.delay.max.ms = 60000
        errors.retry.timeout = 0
        errors.tolerance = none
        header.converter = null
        key.converter = null
        name = card-view-test
        tasks.max = 1
        transforms = []
        value.converter = null
 (org.apache.kafka.connect.runtime.ConnectorConfig$EnrichedConnectorConfig)
[2020-02-16 20:31:46,610] INFO TaskConfig values:
        task.class = class io.confluent.connect.jdbc.source.JdbcSourceTask
 (org.apache.kafka.connect.runtime.TaskConfig)
[2020-02-16 20:31:46,610] INFO Instantiated task card-view-test-0 with version 5.4.0 of type io.confluent.connect.jdbc.source.JdbcSourceTask (org.apache.kafka.connect.runtime.Worker)
[2020-02-16 20:31:46,610] INFO Instantiated connector card-view-test with version 5.4.0 of type class io.confluent.connect.jdbc.JdbcSourceConnector (org.apache.kafka.connect.runtime.Worker)
[2020-02-16 20:31:46,611] INFO Starting JDBC Source Connector (io.confluent.connect.jdbc.JdbcSourceConnector)
[2020-02-16 20:31:46,611] INFO JdbcSourceConnectorConfig values:
        batch.max.rows = 100
        catalog.pattern = null
        connection.attempts = 3
        connection.backoff.ms = 10000
        connection.password = null
        connection.url = jdbc:mysql://10.xx.yy.z:3306/demo_db?user=user&password=pass
        connection.user = null
        db.timezone = UTC
        dialect.name =
        incrementing.column.name = eventId
        mode = incrementing
        numeric.mapping = null
        numeric.precision.mapping = false
        poll.interval.ms = 5000
        query = SELECT * FROM( SELECT HSC.id eventId, HSC.actions, HSC.cause, HSC.metaData, CRED.uid uidCard, CRED.is_primary isPrimary, CRED.type cardType, PR.keys_ prin, HOUSE.uid orgUid, CHILD.uid childUid, USER.first_name fName, USER.last_name lName, CHILD.biz_phone bizPh, CHILD.cell_phone cellPh, AOF.state aof_state, RECORD.uid replacedCardUid FROM house_of_cards as HSC INNER JOIN play_card as CRED ON HSC.debit_card_id = CRED.id INNER JOIN prin PR ON CRED.prin_id = PR.id INNER JOIN child CHILD ON CRED.employee_id = CHILD.id INNER JOIN user USER ON USER.id = CHILD.user_id INNER JOIN house HOUSE ON CHILD.house_id = HOUSE.id LEFT JOIN address SHPADDR ON CRED.shipping_address = SHPADDR.id LEFT JOIN address AOF ON CRED.address_on_file = AOF.id LEFT JOIN play_card RECORD ON CRED.replaced_card_id = RECORD.id WHERE HSC.type = 'cardActionKamen' ) playcards
        quote.sql.identifiers = ALWAYS
        schema.pattern = null
        table.blacklist = []
        table.poll.interval.ms = 60000
        table.types = [TABLE]
        table.whitelist = []
        tables = []
        timestamp.column.name = []
        timestamp.delay.interval.ms = 0
        topic.prefix = card-view-test-25
        validate.non.null = false
 (io.confluent.connect.jdbc.source.JdbcSourceTaskConfig)
[2020-02-16 20:31:46,621] INFO Using JDBC dialect MySql (io.confluent.connect.jdbc.source.JdbcSourceTask)
[2020-02-16 20:31:46,681] INFO Started JDBC source task (io.confluent.connect.jdbc.source.JdbcSourceTask)
[2020-02-16 20:31:46,681] INFO WorkerSourceTask{id=card-view-test-0} Source task finished initialization and start (org.apache.kafka.connect.runtime.WorkerSourceTask)
[2020-02-16 20:31:46,681] INFO Attempting to open connection #1 to MySql (io.confluent.connect.jdbc.util.CachedConnectionProvider)
[2020-02-16 20:31:46,723] INFO [Producer clientId=connector-producer-card-view-test-0] Cluster ID: XXOJHDDIEHID-YYGDIEDi82eeh (org.apache.kafka.clients.Metadata)
[2020-02-16 20:31:46,732] INFO Starting thread to monitor tables. (io.confluent.connect.jdbc.source.TableMonitorThread)
[2020-02-16 20:31:46,735] INFO Finished creating connector card-view-test (org.apache.kafka.connect.runtime.Worker)
[2020-02-16 20:31:46,736] INFO SourceConnectorConfig values:
        config.action.reload = restart
        connector.class = io.confluent.connect.jdbc.JdbcSourceConnector
        errors.log.enable = false
        errors.log.include.messages = false
        errors.retry.delay.max.ms = 60000
        errors.retry.timeout = 0
        errors.tolerance = none
        header.converter = null
        key.converter = null
        name = card-view-test
        tasks.max = 1
        transforms = []
        value.converter = null
 (org.apache.kafka.connect.runtime.SourceConnectorConfig)
[2020-02-16 20:31:46,737] INFO EnrichedConnectorConfig values:
        config.action.reload = restart
        connector.class = io.confluent.connect.jdbc.JdbcSourceConnector
        errors.log.enable = false
        errors.log.include.messages = false
        errors.retry.delay.max.ms = 60000
        errors.retry.timeout = 0
        errors.tolerance = none
        header.converter = null
        key.converter = null
        name = card-view-test
        tasks.max = 1
        transforms = []
        value.converter = null
 (org.apache.kafka.connect.runtime.ConnectorConfig$EnrichedConnectorConfig)
[2020-02-16 20:31:46,949] INFO Begin using SQL query: SELECT * FROM( SELECT HSC.id eventId, HSC.actions, HSC.cause, HSC.metaData, CRED.uid uidCard,
 CRED.is_primary isPrimary, CRED.type cardType, PR.keys_ prin, HOUSE.uid orgUid, CHILD.uid childUid, USER.first_name fName,
 USER.last_name lName, CHILD.biz_phone bizPh, CHILD.cell_phone cellPh, AOF.state aof_state, RECORD.uid replacedCardUid
 FROM house_of_cards as HSC INNER JOIN play_card as CRED ON HSC.debit_card_id = CRED.id INNER JOIN prin PR ON CRED.prin_id = PR.id
 INNER JOIN child CHILD ON CRED.employee_id = CHILD.id INNER JOIN user USER ON USER.id = CHILD.user_id INNER JOIN house HOUSE ON CHILD.house_id = HOUSE.id
 LEFT JOIN address SHPADDR ON CRED.shipping_address = SHPADDR.id LEFT JOIN address AOF ON CRED.address_on_file = AOF.id
 LEFT JOIN play_card RECORD ON CRED.replaced_card_id = RECORD.id WHERE HSC.type = 'cardActionKamen' ) playcards events WHERE `eventId` > ? ORDER BY `eventId` ASC (io.confluent.connect.jdbc.source.TableQuerier)

[2020-02-16 20:31:46,954] ERROR Failed to run query for table TimestampIncrementingTableQuerier{table=null, query='SELECT * FROM( SELECT HSC.id eventId, HSC.actions, HSC.cause, HSC.metaData, CRED.uid uidCard,
 CRED.is_primary isPrimary, CRED.type cardType, PR.keys_ prin, HOUSE.uid orgUid, CHILD.uid childUid, USER.first_name fName,
 USER.last_name lName, CHILD.biz_phone bizPh, CHILD.cell_phone cellPh, AOF.state aof_state, RECORD.uid replacedCardUid
 FROM house_of_cards as HSC INNER JOIN play_card as CRED ON HSC.debit_card_id = CRED.id INNER JOIN prin PR ON CRED.prin_id = PR.id
 INNER JOIN child CHILD ON CRED.employee_id = CHILD.id INNER JOIN user USER ON USER.id = CHILD.user_id INNER JOIN house HOUSE ON CHILD.house_id = HOUSE.id
 LEFT JOIN address SHPADDR ON CRED.shipping_address = SHPADDR.id LEFT JOIN address AOF ON CRED.address_on_file = AOF.id
 LEFT JOIN play_card RECORD ON CRED.replaced_card_id = RECORD.id WHERE HSC.type = "cardActionKamen" )', topicPrefix='card-view-test-25', incrementingColumn='eventId', timestampColumns=[]
}: {} (io.confluent.connect.jdbc.source.JdbcSourceTask)

com.mysql.jdbc.exceptions.jdbc4.MySQLSyntaxErrorException: Table 'demo_db.HSC' doesn't exist
        at sun.reflect.GeneratedConstructorAccessor48.newInstance(Unknown Source)
        at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
        at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
        at com.mysql.jdbc.Util.handleNewInstance(Util.java:404)
        at com.mysql.jdbc.Util.getInstance(Util.java:387)
        at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:942)
        at com.mysql.jdbc.MysqlIO.checkErrorPacket(MysqlIO.java:3966)
        at com.mysql.jdbc.MysqlIO.checkErrorPacket(MysqlIO.java:3902)
        at com.mysql.jdbc.MysqlIO.sendCommand(MysqlIO.java:2526)
        at com.mysql.jdbc.MysqlIO.sqlQueryDirect(MysqlIO.java:2673)
        at com.mysql.jdbc.ConnectionImpl.execSQL(ConnectionImpl.java:2545)
        at com.mysql.jdbc.ConnectionImpl.execSQL(ConnectionImpl.java:2503)
        at com.mysql.jdbc.StatementImpl.executeQuery(StatementImpl.java:1369)
        at com.mysql.jdbc.Field.getCollation(Field.java:448)
        at com.mysql.jdbc.ResultSetMetaData.isCaseSensitive(ResultSetMetaData.java:552)
        at io.confluent.connect.jdbc.dialect.GenericDatabaseDialect.describeColumn(GenericDatabaseDialect.java:713)
        at io.confluent.connect.jdbc.dialect.GenericDatabaseDialect.describeColumns(GenericDatabaseDialect.java:655)
        at io.confluent.connect.jdbc.source.SchemaMapping.create(SchemaMapping.java:63)
        at io.confluent.connect.jdbc.source.TableQuerier.maybeStartQuery(TableQuerier.java:94)
        at io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier.maybeStartQuery(TimestampIncrementingTableQuerier.java:61)
        at io.confluent.connect.jdbc.source.JdbcSourceTask.poll(JdbcSourceTask.java:315)
        at org.apache.kafka.connect.runtime.WorkerSourceTask.poll(WorkerSourceTask.java:265)
        at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:232)
        at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:177)
        at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:227)
        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:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
[2020-02-16 20:31:47,109] INFO [Worker clientId=connect-1, groupId=alpha-core-kafka-connect-wrapper] Tasks [card-view-test-0] configs updated (org.apache.kafka.connect.runtime.distributed.DistributedHerder)
[2020-02-16 20:31:47,611] INFO [Worker clientId=connect-1, groupId=alpha-core-kafka-connect-wrapper] Finished starting connectors and tasks (org.apache.kafka.connect.runtime.distributed.DistributedHerder)
[2020-02-16 20:31:47,613] INFO [Worker clientId=connect-1, groupId=alpha-core-kafka-connect-wrapper] Handling task config update by restarting tasks [card-view-test-0] (org.apache.kafka.connect.runtime.distributed.Distribute
dHerder)
[2020-02-16 20:31:47,613] INFO Stopping task card-view-test-0 (org.apache.kafka.connect.runtime.Worker)
[2020-02-16 20:31:47,614] INFO Stopping JDBC source task (io.confluent.connect.jdbc.source.JdbcSourceTask)
[2020-02-16 20:31:47,656] INFO Closing resources for JDBC source task (io.confluent.connect.jdbc.source.JdbcSourceTask)
[2020-02-16 20:31:47,656] INFO Closing connection #1 to MySql (io.confluent.connect.jdbc.util.CachedConnectionProvider)
[2020-02-16 20:31:47,657] INFO WorkerSourceTask{id=card-view-test-0} Committing offsets (org.apache.kafka.connect.runtime.WorkerSourceTask)
[2020-02-16 20:31:47,657] INFO WorkerSourceTask{id=card-view-test-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask)
[2020-02-16 20:31:47,657] INFO [Producer clientId=connector-producer-card-view-test-0] Closing the Kafka producer with timeoutMillis = 30000 ms. (org.apache.kafka.clients.producer.KafkaProducer)

```

<p><strong><em>EDIT 2. Adding connetor config</em></strong></p>

```{
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "mode": "incrementing",
    "incrementing.column.name": "eventId",
    "topic.prefix": "test-connector",
    "validate.non.null": "false",
    "query": "SELECT * FROM( SELECT HSC.id eventId, HSC.actions, HSC.cause, HSC.metaData, CRED.uid uidCard, CRED.is_primary isPrimary, CRED.type cardType, PR.keys_ prin, HOUSE.uid orgUid, CHILD.uid childUid, USER.first_name fName, USER.last_name lName, CHILD.biz_phone bizPh, CHILD.cell_phone cellPh, AOF.state aof_state, RECORD.uid replacedCardUid FROM house_of_cards as HSC INNER JOIN play_card as CRED ON HSC.debit_card_id = CRED.id INNER JOIN prin PR ON CRED.prin_id = PR.id INNER JOIN child CHILD ON CRED.employee_id = CHILD.id INNER JOIN user USER ON USER.id = CHILD.user_id INNER JOIN house HOUSE ON CHILD.house_id = HOUSE.id LEFT JOIN address SHPADDR ON CRED.shipping_address = SHPADDR.id LEFT JOIN address AOF ON CRED.address_on_file = AOF.id LEFT JOIN play_card RECORD ON CRED.replaced_card_id = RECORD.id WHERE HSC.type = 'cardActionKamen' ) playcards",
    "name": "test-connector-",
    "connection.url": "dbc:mysql://10.xx.yy.z:3306/demo_db?user=user&password=pass"
}
```

Both the queries have WHERE clause, use aliases, have JOINS, etc. Their functionalities seem identical to me, how come one works and the other doesn't? Is there any limit on the length of the query? Is this related to the version of Mysql (I am using 5.7 provided by GCP) or the connector version (I have downloaded 5.1.39 as you can see in the above Dockerfile)?

I have also tried creating a MySQL view of both the subqueries (the sub-SELECT queries) mentioned in the above queries and that works okay for both the queries. Any idea of what might be the issue here?

答:
I tried to reproduce your case.

Create tables in demo_db:

```<code>mysql> show tables;
+-------------------+
| Tables_in_demo_db |
+-------------------+
| address           |
| child             |
| house             |
| house_of_cards    |
| play_card         |
| prin              |
| user              |
+-------------------+
7 rows in set (0.00 sec)
</code>```

Connector launch:

```<code>curl --location --request PUT 'http://myhost:10900/connectors/test-connector/config' \
--header 'Accept: application/json' \
--header 'Content-Type: application/json' \
--data-raw '{
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "mode": "incrementing",
    "incrementing.column.name": "eventId",
    "topic.prefix": "test-connector",
    "validate.non.null": "false",
    "query": "SELECT * FROM( SELECT HSC.id eventId, HSC.actions, HSC.cause, HSC.metaData, CRED.uid uidCard, CRED.is_primary isPrimary, CRED.type cardType, PR.keys_ prin, HOUSE.uid orgUid, CHILD.uid childUid, USER.first_name fName, USER.last_name lName, CHILD.biz_phone bizPh, CHILD.cell_phone cellPh, AOF.state aof_state, RECORD.uid replacedCardUid FROM house_of_cards as HSC INNER JOIN play_card as CRED ON HSC.debit_card_id = CRED.id INNER JOIN prin PR ON CRED.prin_id = PR.id INNER JOIN child CHILD ON CRED.employee_id = CHILD.id INNER JOIN user USER ON USER.id = CHILD.user_id INNER JOIN house HOUSE ON CHILD.house_id = HOUSE.id LEFT JOIN address SHPADDR ON CRED.shipping_address = SHPADDR.id LEFT JOIN address AOF ON CRED.address_on_file = AOF.id LEFT JOIN play_card RECORD ON CRED.replaced_card_id = RECORD.id WHERE HSC.type = '\''cardActionKamen'\'' ) playcards",
    "connection.url": "jdbc:mysql://myhost:3306/demo_db?user=isk&password=123"
}'
</code>```

Connector logs:

```<code>[2020-02-17 02:17:17,589] INFO [test-connector|worker] Creating connector test-connector of type io.confluent.connect.jdbc.JdbcSourceConnector (org.apache.kafka.connect.runtime.Worker:251)
[2020-02-17 02:17:17,592] INFO [test-connector|worker] Instantiated connector test-connector with version 5.3.1 of type class io.confluent.connect.jdbc.JdbcSourceConnector (org.apache.kafka.connect.runtime.Worker:254)
[2020-02-17 02:17:17,592] INFO [test-connector|worker] Starting JDBC Source Connector (io.confluent.connect.jdbc.JdbcSourceConnector:69)
[2020-02-17 02:17:17,592] INFO [test-connector|worker] JdbcSourceConnectorConfig values:
        batch.max.rows = 100
        catalog.pattern = null
        connection.attempts = 3
        connection.backoff.ms = 10000
        connection.password = null
        connection.url = jdbc:mysql://myhost:3306/demo_db?user=isk&password=123
        connection.user = null
        db.timezone = UTC
        dialect.name =
        incrementing.column.name = eventId
        mode = incrementing
        numeric.mapping = null
        numeric.precision.mapping = false
        poll.interval.ms = 5000
        query = SELECT * FROM( SELECT HSC.id eventId, HSC.actions, HSC.cause, HSC.metaData, CRED.uid uidCard, CRED.is_primary isPrimary, CRED.type cardType, PR.keys_ prin, HOUSE.uid orgUid, CHILD.uid childUid, USER.first_name fName, USER.last_name lName, CHILD.biz_phone bizPh, CHILD.cell_phone cellPh, AOF.state aof_state, RECORD.uid replacedCardUid FROM house_of_cards as HSC INNER JOIN play_card as CRED ON HSC.debit_card_id = CRED.id INNER JOIN prin PR ON CRED.prin_id = PR.id INNER JOIN child CHILD ON CRED.employee_id = CHILD.id INNER JOIN user USER ON USER.id = CHILD.user_id INNER JOIN house HOUSE ON CHILD.house_id = HOUSE.id LEFT JOIN address SHPADDR ON CRED.shipping_address = SHPADDR.id LEFT JOIN address AOF ON CRED.address_on_file = AOF.id LEFT JOIN play_card RECORD ON CRED.replaced_card_id = RECORD.id WHERE HSC.type = 'cardActionKamen' ) playcards
        quote.sql.identifiers = ALWAYS
        schema.pattern = null
        table.blacklist = []
        table.poll.interval.ms = 60000
        table.types = [TABLE]
        table.whitelist = []
        timestamp.column.name = []
        timestamp.delay.interval.ms = 0
        topic.prefix = test-connector
        validate.non.null = false
 (io.confluent.connect.jdbc.source.JdbcSourceConnectorConfig:347)
[2020-02-17 02:17:17,594] INFO [test-connector|worker] Attempting to open connection #1 to MySql (io.confluent.connect.jdbc.util.CachedConnectionProvider:87)
[2020-02-17 02:17:17,640] INFO [test-connector|worker] Starting thread to monitor tables. (io.confluent.connect.jdbc.source.TableMonitorThread:73)
[2020-02-17 02:17:17,641] INFO [test-connector|worker] Finished creating connector test-connector (org.apache.kafka.connect.runtime.Worker:273)
</code>```

So there is no error and everything works fine.

Used versions:

Kafka Connect:

```<code>curl -X GET http://localhost:10900 | jq '.version'

"2.4.0"
</code>```

Jdbc Source Connector:

```<code>curl -X GET http://localhost:10900/connector-plugins | jq -c '.[] | select( .class == "io.confluent.connect.jdbc.JdbcSourceConnector") | .version'

"5.3.1"
</code>```

MySql jdbc driver:

```<code>8.0.19
</code>```

Title: MySql query fails in Kafka-connect

Body:
I am using Kafka connect (confluentinc/cp-kafka-connect:5.4.0) and have MySQL connector installed in it. Basically the following Dockerfile:

FROM confluentinc/cp-kafka-connect:5.4.0

RUN echo "===> Installing MySQL connector" \
    && curl -k -SL "https://dev.mysql.com/get/Downloads/Connector-J/mysql-connector-java-5.1.39.tar.gz" | tar -xzf - -C /usr/share/java/kafka-connect-jdbc/ --strip-components=1 mysql-connector-java-5.1.39/mysql-connector-java-5.1.39-bin.jar

I created a connector with the following configurations:

curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
        "name": "test-connector",
        "config": {
                "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
                "connection.url":"jdbc:mysql://10.xx.yy.z:3306/demo_db?user=user&password=pass",
                "topic.prefix": "test-connector-",
                "mode":"incrementing",
                "query":"$QUERY",
                "incrementing.column.name": "eventId",
                "validate.non.null": false
                }
        }'

I tried replacing the "$QUERY" mentioned above with the following query (pasting a formatted version for better readability):

SELECT * FROM (
    select DISTINCT t1.id,
            t1.name,
            t1.email,
            t1.department,
            t1.modified
        from
            test as t1
            LEFT OUTER JOIN test as t2 ON t1.id > -1
        WHERE
            t1.id > -1) something

The above connector gets created and works as expected. However, the below query fails. and

SELECT * FROM(
        SELECT
            HSC.id eventId,
            HSC.actions,
            HSC.cause,
            HSC.metaData,
            CRED.uid uidCard,
            CRED.is_primary isPrimary,
            CRED.type cardType,
            PR.keys_ prin,
            HOUSE.uid orgUid,
            CHILD.uid childUid,
            USER.first_name fName,
            USER.last_name lName,
            CHILD.biz_phone bizPh,
            CHILD.cell_phone cellPh,
            AOF.state aof_state,
            RECORD.uid replacedCardUid
        FROM
            house_of_cards as HSC
            INNER JOIN play_card as CRED ON HSC.debit_card_id = CRED.id
            INNER JOIN prin PR ON CRED.prin_id = PR.id
            INNER JOIN child CHILD ON CRED.employee_id = CHILD.id
            INNER JOIN user USER ON USER.id = CHILD.user_id
            INNER JOIN house HOUSE ON CHILD.house_id = HOUSE.id
            LEFT JOIN address SHPADDR ON CRED.shipping_address = SHPADDR.id
            LEFT JOIN address AOF ON CRED.address_on_file = AOF.id
            LEFT JOIN play_card RECORD ON CRED.replaced_card_id = RECORD.id
        WHERE
        HSC.type = 'cardActionKamen' ) playcards

EDIT: 1. Posting full error log  (with modified sensitive data of course)

[2020-02-16 20:31:46,463] INFO AbstractConfig values:
        batch.max.rows = 100
        catalog.pattern = null
        connection.attempts = 3
        connection.backoff.ms = 10000
        connection.password = null
        connection.url = jdbc:mysql://10.xx.yy.z:3306/demo_db?user=user&password=pass
        connection.user = null
        db.timezone = UTC
        dialect.name =
        incrementing.column.name = eventId
        mode = incrementing
        numeric.mapping = null
        numeric.precision.mapping = false
        poll.interval.ms = 5000
        query = SELECT * FROM( SELECT HSC.id eventId, HSC.actions, HSC.cause, HSC.metaData, CRED.uid uidCard, CRED.is_primary isPrimary, CRED.type cardType, PR.keys_ prin, HOUSE.uid orgUid, CHILD.uid childUid, USER.first_name fName, USER.last_name lName, CHILD.biz_phone bizPh, CHILD.cell_phone cellPh, AOF.state aof_state, RECORD.uid replacedCardUid FROM house_of_cards as HSC INNER JOIN play_card as CRED ON HSC.debit_card_id = CRED.id INNER JOIN prin PR ON CRED.prin_id = PR.id INNER JOIN child CHILD ON CRED.employee_id = CHILD.id INNER JOIN user USER ON USER.id = CHILD.user_id INNER JOIN house HOUSE ON CHILD.house_id = HOUSE.id LEFT JOIN address SHPADDR ON CRED.shipping_address = SHPADDR.id LEFT JOIN address AOF ON CRED.address_on_file = AOF.id LEFT JOIN play_card RECORD ON CRED.replaced_card_id = RECORD.id WHERE HSC.type = 'cardActionKamen' ) playcards
        quote.sql.identifiers = ALWAYS
        schema.pattern = null
        table.blacklist = []
        table.poll.interval.ms = 60000
        table.types = [TABLE]
        table.whitelist = []
        timestamp.column.name = []
        timestamp.delay.interval.ms = 0
        topic.prefix = card-view-test
        validate.non.null = false
 (org.apache.kafka.common.config.AbstractConfig)
[2020-02-16 20:31:46,595] INFO AbstractConfig values:
 (org.apache.kafka.common.config.AbstractConfig)
[2020-02-16 20:31:46,601] INFO [Worker clientId=connect-1, groupId=alpha-core-kafka-connect-wrapper] Connector card-view-test config updated (org.apache.kafka.connect.runtime.distributed.DistributedHerder)
[2020-02-16 20:31:46,602] INFO [Worker clientId=connect-1, groupId=alpha-core-kafka-connect-wrapper] Rebalance started (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator)
[2020-02-16 20:31:46,602] INFO [Worker clientId=connect-1, groupId=alpha-core-kafka-connect-wrapper] (Re-)joining group (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
[2020-02-16 20:31:46,609] INFO [Worker clientId=connect-1, groupId=alpha-core-kafka-connect-wrapper] Successfully joined group with generation 295 (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
[2020-02-16 20:31:46,609] INFO [Worker clientId=connect-1, groupId=alpha-core-kafka-connect-wrapper] Joined group at generation 295 with protocol version 2 and got assignment: Assignment{error=0, leader='connect-1-d82cc98e-5
619-4972-ad78-3b0a53c3b5bb', leaderUrl='http://10.xx.yy.36:8083/', offset=433, connectorIds=[card-view-test], taskIds=[card-view-test-0], revokedConnectorIds=[], revokedTaskIds=[], delay=0} with rebalance delay: 0 (org.apache
.kafka.connect.runtime.distributed.DistributedHerder)
[2020-02-16 20:31:46,609] INFO [Worker clientId=connect-1, groupId=alpha-core-kafka-connect-wrapper] Starting connectors and tasks using config offset 433 (org.apache.kafka.connect.runtime.distributed.DistributedHerder)
[2020-02-16 20:31:46,609] INFO [Worker clientId=connect-1, groupId=alpha-core-kafka-connect-wrapper] Starting connector card-view-test (org.apache.kafka.connect.runtime.distributed.DistributedHerder)
[2020-02-16 20:31:46,609] INFO [Worker clientId=connect-1, groupId=alpha-core-kafka-connect-wrapper] Starting task card-view-test-0 (org.apache.kafka.connect.runtime.distributed.DistributedHerder)
[2020-02-16 20:31:46,610] INFO Creating task card-view-test-0 (org.apache.kafka.connect.runtime.Worker)
[2020-02-16 20:31:46,610] INFO ConnectorConfig values:
        config.action.reload = restart
        connector.class = io.confluent.connect.jdbc.JdbcSourceConnector
        errors.log.enable = false
        errors.log.include.messages = false
        errors.retry.delay.max.ms = 60000
        errors.retry.timeout = 0
        errors.tolerance = none
        header.converter = null
        key.converter = null
        name = card-view-test
        tasks.max = 1
        transforms = []
        value.converter = null
 (org.apache.kafka.connect.runtime.ConnectorConfig)
[2020-02-16 20:31:46,610] INFO EnrichedConnectorConfig values:
        config.action.reload = restart
        connector.class = io.confluent.connect.jdbc.JdbcSourceConnector
        errors.log.enable = false
        errors.log.include.messages = false
        errors.retry.delay.max.ms = 60000
        errors.retry.timeout = 0
        errors.tolerance = none
        header.converter = null
        key.converter = null
        name = card-view-test
        tasks.max = 1
        transforms = []
        value.converter = null
 (org.apache.kafka.connect.runtime.ConnectorConfig$EnrichedConnectorConfig)
[2020-02-16 20:31:46,610] INFO Creating connector card-view-test of type io.confluent.connect.jdbc.JdbcSourceConnector (org.apache.kafka.connect.runtime.Worker)
[2020-02-16 20:31:46,610] INFO ConnectorConfig values:
        config.action.reload = restart
        connector.class = io.confluent.connect.jdbc.JdbcSourceConnector
        errors.log.enable = false
        errors.log.include.messages = false
        errors.retry.delay.max.ms = 60000
        errors.retry.timeout = 0
        errors.tolerance = none
        header.converter = null
        key.converter = null
        name = card-view-test
        tasks.max = 1
        transforms = []
        value.converter = null
 (org.apache.kafka.connect.runtime.ConnectorConfig)
[2020-02-16 20:31:46,610] INFO EnrichedConnectorConfig values:
        config.action.reload = restart
        connector.class = io.confluent.connect.jdbc.JdbcSourceConnector
        errors.log.enable = false
        errors.log.include.messages = false
        errors.retry.delay.max.ms = 60000
        errors.retry.timeout = 0
        errors.tolerance = none
        header.converter = null
        key.converter = null
        name = card-view-test
        tasks.max = 1
        transforms = []
        value.converter = null
 (org.apache.kafka.connect.runtime.ConnectorConfig)
[2020-02-16 20:31:46,610] INFO EnrichedConnectorConfig values:
        config.action.reload = restart
        connector.class = io.confluent.connect.jdbc.JdbcSourceConnector
        errors.log.enable = false
        errors.log.include.messages = false
        errors.retry.delay.max.ms = 60000
        errors.retry.timeout = 0
        errors.tolerance = none
        header.converter = null
        key.converter = null
        name = card-view-test
        tasks.max = 1
        transforms = []
        value.converter = null
 (org.apache.kafka.connect.runtime.ConnectorConfig$EnrichedConnectorConfig)
[2020-02-16 20:31:46,610] INFO TaskConfig values:
        task.class = class io.confluent.connect.jdbc.source.JdbcSourceTask
 (org.apache.kafka.connect.runtime.TaskConfig)
[2020-02-16 20:31:46,610] INFO Instantiated task card-view-test-0 with version 5.4.0 of type io.confluent.connect.jdbc.source.JdbcSourceTask (org.apache.kafka.connect.runtime.Worker)
[2020-02-16 20:31:46,610] INFO Instantiated connector card-view-test with version 5.4.0 of type class io.confluent.connect.jdbc.JdbcSourceConnector (org.apache.kafka.connect.runtime.Worker)
[2020-02-16 20:31:46,611] INFO Starting JDBC Source Connector (io.confluent.connect.jdbc.JdbcSourceConnector)
[2020-02-16 20:31:46,611] INFO JdbcSourceConnectorConfig values:
        batch.max.rows = 100
        catalog.pattern = null
        connection.attempts = 3
        connection.backoff.ms = 10000
        connection.password = null
        connection.url = jdbc:mysql://10.xx.yy.z:3306/demo_db?user=user&password=pass
        connection.user = null
        db.timezone = UTC
        dialect.name =
        incrementing.column.name = eventId
        mode = incrementing
        numeric.mapping = null
        numeric.precision.mapping = false
        poll.interval.ms = 5000
        query = SELECT * FROM( SELECT HSC.id eventId, HSC.actions, HSC.cause, HSC.metaData, CRED.uid uidCard, CRED.is_primary isPrimary, CRED.type cardType, PR.keys_ prin, HOUSE.uid orgUid, CHILD.uid childUid, USER.first_name fName, USER.last_name lName, CHILD.biz_phone bizPh, CHILD.cell_phone cellPh, AOF.state aof_state, RECORD.uid replacedCardUid FROM house_of_cards as HSC INNER JOIN play_card as CRED ON HSC.debit_card_id = CRED.id INNER JOIN prin PR ON CRED.prin_id = PR.id INNER JOIN child CHILD ON CRED.employee_id = CHILD.id INNER JOIN user USER ON USER.id = CHILD.user_id INNER JOIN house HOUSE ON CHILD.house_id = HOUSE.id LEFT JOIN address SHPADDR ON CRED.shipping_address = SHPADDR.id LEFT JOIN address AOF ON CRED.address_on_file = AOF.id LEFT JOIN play_card RECORD ON CRED.replaced_card_id = RECORD.id WHERE HSC.type = 'cardActionKamen' ) playcards
        quote.sql.identifiers = ALWAYS
        schema.pattern = null
        table.blacklist = []
        table.poll.interval.ms = 60000
        table.types = [TABLE]
        table.whitelist = []
        tables = []
        timestamp.column.name = []
        timestamp.delay.interval.ms = 0
        topic.prefix = card-view-test-25
        validate.non.null = false
 (io.confluent.connect.jdbc.source.JdbcSourceTaskConfig)
[2020-02-16 20:31:46,621] INFO Using JDBC dialect MySql (io.confluent.connect.jdbc.source.JdbcSourceTask)
[2020-02-16 20:31:46,681] INFO Started JDBC source task (io.confluent.connect.jdbc.source.JdbcSourceTask)
[2020-02-16 20:31:46,681] INFO WorkerSourceTask{id=card-view-test-0} Source task finished initialization and start (org.apache.kafka.connect.runtime.WorkerSourceTask)
[2020-02-16 20:31:46,681] INFO Attempting to open connection #1 to MySql (io.confluent.connect.jdbc.util.CachedConnectionProvider)
[2020-02-16 20:31:46,723] INFO [Producer clientId=connector-producer-card-view-test-0] Cluster ID: XXOJHDDIEHID-YYGDIEDi82eeh (org.apache.kafka.clients.Metadata)
[2020-02-16 20:31:46,732] INFO Starting thread to monitor tables. (io.confluent.connect.jdbc.source.TableMonitorThread)
[2020-02-16 20:31:46,735] INFO Finished creating connector card-view-test (org.apache.kafka.connect.runtime.Worker)
[2020-02-16 20:31:46,736] INFO SourceConnectorConfig values:
        config.action.reload = restart
        connector.class = io.confluent.connect.jdbc.JdbcSourceConnector
        errors.log.enable = false
        errors.log.include.messages = false
        errors.retry.delay.max.ms = 60000
        errors.retry.timeout = 0
        errors.tolerance = none
        header.converter = null
        key.converter = null
        name = card-view-test
        tasks.max = 1
        transforms = []
        value.converter = null
 (org.apache.kafka.connect.runtime.SourceConnectorConfig)
[2020-02-16 20:31:46,737] INFO EnrichedConnectorConfig values:
        config.action.reload = restart
        connector.class = io.confluent.connect.jdbc.JdbcSourceConnector
        errors.log.enable = false
        errors.log.include.messages = false
        errors.retry.delay.max.ms = 60000
        errors.retry.timeout = 0
        errors.tolerance = none
        header.converter = null
        key.converter = null
        name = card-view-test
        tasks.max = 1
        transforms = []
        value.converter = null
 (org.apache.kafka.connect.runtime.ConnectorConfig$EnrichedConnectorConfig)
[2020-02-16 20:31:46,949] INFO Begin using SQL query: SELECT * FROM( SELECT HSC.id eventId, HSC.actions, HSC.cause, HSC.metaData, CRED.uid uidCard,
 CRED.is_primary isPrimary, CRED.type cardType, PR.keys_ prin, HOUSE.uid orgUid, CHILD.uid childUid, USER.first_name fName,
 USER.last_name lName, CHILD.biz_phone bizPh, CHILD.cell_phone cellPh, AOF.state aof_state, RECORD.uid replacedCardUid
 FROM house_of_cards as HSC INNER JOIN play_card as CRED ON HSC.debit_card_id = CRED.id INNER JOIN prin PR ON CRED.prin_id = PR.id
 INNER JOIN child CHILD ON CRED.employee_id = CHILD.id INNER JOIN user USER ON USER.id = CHILD.user_id INNER JOIN house HOUSE ON CHILD.house_id = HOUSE.id
 LEFT JOIN address SHPADDR ON CRED.shipping_address = SHPADDR.id LEFT JOIN address AOF ON CRED.address_on_file = AOF.id
 LEFT JOIN play_card RECORD ON CRED.replaced_card_id = RECORD.id WHERE HSC.type = 'cardActionKamen' ) playcards events WHERE `eventId` > ? ORDER BY `eventId` ASC (io.confluent.connect.jdbc.source.TableQuerier)

[2020-02-16 20:31:46,954] ERROR Failed to run query for table TimestampIncrementingTableQuerier{table=null, query='SELECT * FROM( SELECT HSC.id eventId, HSC.actions, HSC.cause, HSC.metaData, CRED.uid uidCard,
 CRED.is_primary isPrimary, CRED.type cardType, PR.keys_ prin, HOUSE.uid orgUid, CHILD.uid childUid, USER.first_name fName,
 USER.last_name lName, CHILD.biz_phone bizPh, CHILD.cell_phone cellPh, AOF.state aof_state, RECORD.uid replacedCardUid
 FROM house_of_cards as HSC INNER JOIN play_card as CRED ON HSC.debit_card_id = CRED.id INNER JOIN prin PR ON CRED.prin_id = PR.id
 INNER JOIN child CHILD ON CRED.employee_id = CHILD.id INNER JOIN user USER ON USER.id = CHILD.user_id INNER JOIN house HOUSE ON CHILD.house_id = HOUSE.id
 LEFT JOIN address SHPADDR ON CRED.shipping_address = SHPADDR.id LEFT JOIN address AOF ON CRED.address_on_file = AOF.id
 LEFT JOIN play_card RECORD ON CRED.replaced_card_id = RECORD.id WHERE HSC.type = "cardActionKamen" )', topicPrefix='card-view-test-25', incrementingColumn='eventId', timestampColumns=[]
}: {} (io.confluent.connect.jdbc.source.JdbcSourceTask)

com.mysql.jdbc.exceptions.jdbc4.MySQLSyntaxErrorException: Table 'demo_db.HSC' doesn't exist
        at sun.reflect.GeneratedConstructorAccessor48.newInstance(Unknown Source)
        at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
        at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
        at com.mysql.jdbc.Util.handleNewInstance(Util.java:404)
        at com.mysql.jdbc.Util.getInstance(Util.java:387)
        at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:942)
        at com.mysql.jdbc.MysqlIO.checkErrorPacket(MysqlIO.java:3966)
        at com.mysql.jdbc.MysqlIO.checkErrorPacket(MysqlIO.java:3902)
        at com.mysql.jdbc.MysqlIO.sendCommand(MysqlIO.java:2526)
        at com.mysql.jdbc.MysqlIO.sqlQueryDirect(MysqlIO.java:2673)
        at com.mysql.jdbc.ConnectionImpl.execSQL(ConnectionImpl.java:2545)
        at com.mysql.jdbc.ConnectionImpl.execSQL(ConnectionImpl.java:2503)
        at com.mysql.jdbc.StatementImpl.executeQuery(StatementImpl.java:1369)
        at com.mysql.jdbc.Field.getCollation(Field.java:448)
        at com.mysql.jdbc.ResultSetMetaData.isCaseSensitive(ResultSetMetaData.java:552)
        at io.confluent.connect.jdbc.dialect.GenericDatabaseDialect.describeColumn(GenericDatabaseDialect.java:713)
        at io.confluent.connect.jdbc.dialect.GenericDatabaseDialect.describeColumns(GenericDatabaseDialect.java:655)
        at io.confluent.connect.jdbc.source.SchemaMapping.create(SchemaMapping.java:63)
        at io.confluent.connect.jdbc.source.TableQuerier.maybeStartQuery(TableQuerier.java:94)
        at io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier.maybeStartQuery(TimestampIncrementingTableQuerier.java:61)
        at io.confluent.connect.jdbc.source.JdbcSourceTask.poll(JdbcSourceTask.java:315)
        at org.apache.kafka.connect.runtime.WorkerSourceTask.poll(WorkerSourceTask.java:265)
        at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:232)
        at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:177)
        at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:227)
        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:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
[2020-02-16 20:31:47,109] INFO [Worker clientId=connect-1, groupId=alpha-core-kafka-connect-wrapper] Tasks [card-view-test-0] configs updated (org.apache.kafka.connect.runtime.distributed.DistributedHerder)
[2020-02-16 20:31:47,611] INFO [Worker clientId=connect-1, groupId=alpha-core-kafka-connect-wrapper] Finished starting connectors and tasks (org.apache.kafka.connect.runtime.distributed.DistributedHerder)
[2020-02-16 20:31:47,613] INFO [Worker clientId=connect-1, groupId=alpha-core-kafka-connect-wrapper] Handling task config update by restarting tasks [card-view-test-0] (org.apache.kafka.connect.runtime.distributed.Distribute
dHerder)
[2020-02-16 20:31:47,613] INFO Stopping task card-view-test-0 (org.apache.kafka.connect.runtime.Worker)
[2020-02-16 20:31:47,614] INFO Stopping JDBC source task (io.confluent.connect.jdbc.source.JdbcSourceTask)
[2020-02-16 20:31:47,656] INFO Closing resources for JDBC source task (io.confluent.connect.jdbc.source.JdbcSourceTask)
[2020-02-16 20:31:47,656] INFO Closing connection #1 to MySql (io.confluent.connect.jdbc.util.CachedConnectionProvider)
[2020-02-16 20:31:47,657] INFO WorkerSourceTask{id=card-view-test-0} Committing offsets (org.apache.kafka.connect.runtime.WorkerSourceTask)
[2020-02-16 20:31:47,657] INFO WorkerSourceTask{id=card-view-test-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask)
[2020-02-16 20:31:47,657] INFO [Producer clientId=connector-producer-card-view-test-0] Closing the Kafka producer with timeoutMillis = 30000 ms. (org.apache.kafka.clients.producer.KafkaProducer)

EDIT 2. Adding connetor config

{
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "mode": "incrementing",
    "incrementing.column.name": "eventId",
    "topic.prefix": "test-connector",
    "validate.non.null": "false",
    "query": "SELECT * FROM( SELECT HSC.id eventId, HSC.actions, HSC.cause, HSC.metaData, CRED.uid uidCard, CRED.is_primary isPrimary, CRED.type cardType, PR.keys_ prin, HOUSE.uid orgUid, CHILD.uid childUid, USER.first_name fName, USER.last_name lName, CHILD.biz_phone bizPh, CHILD.cell_phone cellPh, AOF.state aof_state, RECORD.uid replacedCardUid FROM house_of_cards as HSC INNER JOIN play_card as CRED ON HSC.debit_card_id = CRED.id INNER JOIN prin PR ON CRED.prin_id = PR.id INNER JOIN child CHILD ON CRED.employee_id = CHILD.id INNER JOIN user USER ON USER.id = CHILD.user_id INNER JOIN house HOUSE ON CHILD.house_id = HOUSE.id LEFT JOIN address SHPADDR ON CRED.shipping_address = SHPADDR.id LEFT JOIN address AOF ON CRED.address_on_file = AOF.id LEFT JOIN play_card RECORD ON CRED.replaced_card_id = RECORD.id WHERE HSC.type = 'cardActionKamen' ) playcards",
    "name": "test-connector-",
    "connection.url": "dbc:mysql://10.xx.yy.z:3306/demo_db?user=user&password=pass"
}

Both the queries have WHERE clause, use aliases, have JOINS, etc. Their functionalities seem identical to me, how come one works and the other doesn't? Is there any limit on the length of the query? Is this related to the version of Mysql (I am using 5.7 provided by GCP) or the connector version (I have downloaded 5.1.39 as you can see in the above Dockerfile)?

I have also tried creating a MySQL view of both the subqueries (the sub-SELECT queries) mentioned in the above queries and that works okay for both the queries. Any idea of what might be the issue here?

Body:
I tried to reproduce your case.

Create tables in demo_db:

mysql> show tables;
+-------------------+
| Tables_in_demo_db |
+-------------------+
| address           |
| child             |
| house             |
| house_of_cards    |
| play_card         |
| prin              |
| user              |
+-------------------+
7 rows in set (0.00 sec)

Connector launch:

curl --location --request PUT 'http://myhost:10900/connectors/test-connector/config' \
--header 'Accept: application/json' \
--header 'Content-Type: application/json' \
--data-raw '{
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "mode": "incrementing",
    "incrementing.column.name": "eventId",
    "topic.prefix": "test-connector",
    "validate.non.null": "false",
    "query": "SELECT * FROM( SELECT HSC.id eventId, HSC.actions, HSC.cause, HSC.metaData, CRED.uid uidCard, CRED.is_primary isPrimary, CRED.type cardType, PR.keys_ prin, HOUSE.uid orgUid, CHILD.uid childUid, USER.first_name fName, USER.last_name lName, CHILD.biz_phone bizPh, CHILD.cell_phone cellPh, AOF.state aof_state, RECORD.uid replacedCardUid FROM house_of_cards as HSC INNER JOIN play_card as CRED ON HSC.debit_card_id = CRED.id INNER JOIN prin PR ON CRED.prin_id = PR.id INNER JOIN child CHILD ON CRED.employee_id = CHILD.id INNER JOIN user USER ON USER.id = CHILD.user_id INNER JOIN house HOUSE ON CHILD.house_id = HOUSE.id LEFT JOIN address SHPADDR ON CRED.shipping_address = SHPADDR.id LEFT JOIN address AOF ON CRED.address_on_file = AOF.id LEFT JOIN play_card RECORD ON CRED.replaced_card_id = RECORD.id WHERE HSC.type = '\''cardActionKamen'\'' ) playcards",
    "connection.url": "jdbc:mysql://myhost:3306/demo_db?user=isk&password=123"
}'

Connector logs:

[2020-02-17 02:17:17,589] INFO [test-connector|worker] Creating connector test-connector of type io.confluent.connect.jdbc.JdbcSourceConnector (org.apache.kafka.connect.runtime.Worker:251)
[2020-02-17 02:17:17,592] INFO [test-connector|worker] Instantiated connector test-connector with version 5.3.1 of type class io.confluent.connect.jdbc.JdbcSourceConnector (org.apache.kafka.connect.runtime.Worker:254)
[2020-02-17 02:17:17,592] INFO [test-connector|worker] Starting JDBC Source Connector (io.confluent.connect.jdbc.JdbcSourceConnector:69)
[2020-02-17 02:17:17,592] INFO [test-connector|worker] JdbcSourceConnectorConfig values:
        batch.max.rows = 100
        catalog.pattern = null
        connection.attempts = 3
        connection.backoff.ms = 10000
        connection.password = null
        connection.url = jdbc:mysql://myhost:3306/demo_db?user=isk&password=123
        connection.user = null
        db.timezone = UTC
        dialect.name =
        incrementing.column.name = eventId
        mode = incrementing
        numeric.mapping = null
        numeric.precision.mapping = false
        poll.interval.ms = 5000
        query = SELECT * FROM( SELECT HSC.id eventId, HSC.actions, HSC.cause, HSC.metaData, CRED.uid uidCard, CRED.is_primary isPrimary, CRED.type cardType, PR.keys_ prin, HOUSE.uid orgUid, CHILD.uid childUid, USER.first_name fName, USER.last_name lName, CHILD.biz_phone bizPh, CHILD.cell_phone cellPh, AOF.state aof_state, RECORD.uid replacedCardUid FROM house_of_cards as HSC INNER JOIN play_card as CRED ON HSC.debit_card_id = CRED.id INNER JOIN prin PR ON CRED.prin_id = PR.id INNER JOIN child CHILD ON CRED.employee_id = CHILD.id INNER JOIN user USER ON USER.id = CHILD.user_id INNER JOIN house HOUSE ON CHILD.house_id = HOUSE.id LEFT JOIN address SHPADDR ON CRED.shipping_address = SHPADDR.id LEFT JOIN address AOF ON CRED.address_on_file = AOF.id LEFT JOIN play_card RECORD ON CRED.replaced_card_id = RECORD.id WHERE HSC.type = 'cardActionKamen' ) playcards
        quote.sql.identifiers = ALWAYS
        schema.pattern = null
        table.blacklist = []
        table.poll.interval.ms = 60000
        table.types = [TABLE]
        table.whitelist = []
        timestamp.column.name = []
        timestamp.delay.interval.ms = 0
        topic.prefix = test-connector
        validate.non.null = false
 (io.confluent.connect.jdbc.source.JdbcSourceConnectorConfig:347)
[2020-02-17 02:17:17,594] INFO [test-connector|worker] Attempting to open connection #1 to MySql (io.confluent.connect.jdbc.util.CachedConnectionProvider:87)
[2020-02-17 02:17:17,640] INFO [test-connector|worker] Starting thread to monitor tables. (io.confluent.connect.jdbc.source.TableMonitorThread:73)
[2020-02-17 02:17:17,641] INFO [test-connector|worker] Finished creating connector test-connector (org.apache.kafka.connect.runtime.Worker:273)

So there is no error and everything works fine.

Used versions:

Kafka Connect:

curl -X GET http://localhost:10900 | jq '.version'

"2.4.0"

Jdbc Source Connector:

curl -X GET http://localhost:10900/connector-plugins | jq -c '.[] | select( .class == "io.confluent.connect.jdbc.JdbcSourceConnector") | .version'

"5.3.1"

MySql jdbc driver:

8.0.19


来源: stackexchange

问:
How to execute a program with Kotlin and Arrow
I'm trying to learn a bit of Functional Programming using Kotlin and Arrow and in this way I've already read some blogposts like the following one: https://jorgecastillo.dev/kotlin-fp-1-monad-stack, which is good, I've understand the main idea, but when creating a program, I can't figure out how to run it.

Let me be more explicit:

I have the following piece of code:

```<code>typealias EitherIO<A, B> = EitherT<ForIO, A, B>

sealed class UserError(
        val message: String,
        val status: Int
) {
    object AuthenticationError : UserError(HttpStatus.UNAUTHORIZED.reasonPhrase, HttpStatus.UNAUTHORIZED.value())
    object UserNotFound : UserError(HttpStatus.NOT_FOUND.reasonPhrase, HttpStatus.NOT_FOUND.value())
    object InternalServerError : UserError(HttpStatus.INTERNAL_SERVER_ERROR.reasonPhrase, HttpStatus.INTERNAL_SERVER_ERROR.value())
}

@Component
class UserAdapter(
        private val myAccountClient: MyAccountClient
) {
    @Lazy
    @Inject
    lateinit var subscriberRepository: SubscriberRepository

    fun getDomainUser(ssoId: Long): EitherIO<UserError, User?> {
        val io = IO.fx {
            val userResource = getUserResourcesBySsoId(ssoId, myAccountClient).bind()
            userResource.fold(
                    { error -> Either.Left(error) },
                    { success ->
                        Either.right(composeDomainUserWithSubscribers(success, getSubscribersForUserResource(success, subscriberRepository).bind()))
                    })
        }
        return EitherIO(io)
    }

    fun composeDomainUserWithSubscribers(userResource: UserResource, subscribers: Option<Subscribers>): User? {
        return subscribers.map { userResource.toDomainUser(it) }.orNull()
    }
}

private fun getSubscribersForUserResource(userResource: UserResource, subscriberRepository: SubscriberRepository): IO<Option<Subscribers>> {
    return IO {
        val msisdnList = userResource.getMsisdnList()
        Option.invoke(subscriberRepository.findAllByMsisdnInAndDeletedIsFalse(msisdnList).associateBy(Subscriber::msisdn))
    }
}

private fun getUserResourcesBySsoId(ssoId: Long, myAccountClient: MyAccountClient): IO<Either<UserError, UserResource>> {
    return IO {
        val response = myAccountClient.getUserBySsoId(ssoId)
        if (response.isSuccessful) {
            val userResource = JacksonUtils.fromJsonToObject(response.body()?.string()!!, UserResource::class.java)
            Either.Right(userResource)
        } else {
            when (response.code()) {
                401 -> Either.Left(UserError.AuthenticationError)
                404 -> Either.Left(UserError.UserNotFound)
                else -> Either.Left(UserError.InternalServerError)
            }
        }
    }.handleError { Either.Left(UserError.InternalServerError) }
}
</code>```

which, as you can see is accumulating some results into an IO monad. I should run this program using unsafeRunSync() from arrow, but on javadoc it's stated the following: **NOTE** this function is intended for testing, it should never appear in your mainline production code!.
I should mention that I know about unsafeRunAsync, but in my case I want to be synchronous.

Thanks!

答:
Instead of running unsafeRunSync, you should favor unsafeRunAsync.

If you have myFun(): IO and want to run this, then you call myFun().unsafeRunAsync(cb) where cb: (Either<Throwable, A>) -> Unit.

For instance, if your function returns IO<List<Int>> then you can call

```<code>myFun().unsafeRunAsync {  /* it (Either<Throwable, List<Int>>) -> */
  it.fold(
    { Log.e("Foo", "Error! $it") },
    { println(it) })
}
</code>```

This will run the program contained in the IO asynchronously and pass the result safely to the callback, which will log an error if the IO threw, and otherwise it will print the list of integers.

You should avoid unsafeRunSync for a number of reasons, discussed here. It's blocking, it can cause crashes, it can cause deadlocks, and it can halt your application.

If you really want to run your IO as a blocking computation, then you can precede this with attempt() to have your IO become an IO<Either<Throwable, A>> similar to the unsafeRunAsync callback parameter. At least then you won't crash.

But unsafeRunAsync is preferred. Also, make sure your callback passed to unsafeRunAsync won't throw any errors, at it's assumed it won't. Docs.

ContentLicense: CC BY-SA 4.0

Body:
I'm trying to learn a bit of Functional Programming using Kotlin and Arrow and in this way I've already read some blogposts like the following one: https://jorgecastillo.dev/kotlin-fp-1-monad-stack, which is good, I've understand the main idea, but when creating a program, I can't figure out how to run it.

Let me be more explicit:

I have the following piece of code:

typealias EitherIO = EitherT<ForIO, A, B>

sealed class UserError(
        val message: String,
        val status: Int
) {
    object AuthenticationError : UserError(HttpStatus.UNAUTHORIZED.reasonPhrase, HttpStatus.UNAUTHORIZED.value())
    object UserNotFound : UserError(HttpStatus.NOT_FOUND.reasonPhrase, HttpStatus.NOT_FOUND.value())
    object InternalServerError : UserError(HttpStatus.INTERNAL_SERVER_ERROR.reasonPhrase, HttpStatus.INTERNAL_SERVER_ERROR.value())
}

@Component
class UserAdapter(
        private val myAccountClient: MyAccountClient
) {
    @Lazy
    @Inject
    lateinit var subscriberRepository: SubscriberRepository

    fun getDomainUser(ssoId: Long): EitherIO<UserError, User?> {
        val io = IO.fx {
            val userResource = getUserResourcesBySsoId(ssoId, myAccountClient).bind()
            userResource.fold(
                    { error -> Either.Left(error) },
                    { success ->
                        Either.right(composeDomainUserWithSubscribers(success, getSubscribersForUserResource(success, subscriberRepository).bind()))
                    })
        }
        return EitherIO(io)
    }

    fun composeDomainUserWithSubscribers(userResource: UserResource, subscribers: Option<Subscribers>): User? {
        return subscribers.map { userResource.toDomainUser(it) }.orNull()
    }
}

private fun getSubscribersForUserResource(userResource: UserResource, subscriberRepository: SubscriberRepository): IO<Option<Subscribers>> {
    return IO {
        val msisdnList = userResource.getMsisdnList()
        Option.invoke(subscriberRepository.findAllByMsisdnInAndDeletedIsFalse(msisdnList).associateBy(Subscriber::msisdn))
    }
}

private fun getUserResourcesBySsoId(ssoId: Long, myAccountClient: MyAccountClient): IO<Either<UserError, UserResource>> {
    return IO {
        val response = myAccountClient.getUserBySsoId(ssoId)
        if (response.isSuccessful) {
            val userResource = JacksonUtils.fromJsonToObject(response.body()?.string()!!, UserResource::class.java)
            Either.Right(userResource)
        } else {
            when (response.code()) {
                401 -> Either.Left(UserError.AuthenticationError)
                404 -> Either.Left(UserError.UserNotFound)
                else -> Either.Left(UserError.InternalServerError)
            }
        }
    }.handleError { Either.Left(UserError.InternalServerError) }
}

which, as you can see is accumulating some results into an IO monad. I should run this program using unsafeRunSync() from arrow, but on javadoc it's stated the following: **NOTE** this function is intended for testing, it should never appear in your mainline production code!.
I should mention that I know about unsafeRunAsync, but in my case I want to be synchronous.

Thanks!

Body:
Instead of running unsafeRunSync, you should favor unsafeRunAsync.

If you have myFun(): IO<A> and want to run this, then you call myFun().unsafeRunAsync(cb) where cb: (Either<Throwable, A>) -> Unit.

For instance, if your function returns IO<List<Int>> then you can call

myFun().unsafeRunAsync {  /* it (Either<Throwable, List<Int>>) -> */
  it.fold(
    { Log.e("Foo", "Error! $it") },
    { println(it) })
}

This will run the program contained in the IO asynchronously and pass the result safely to the callback, which will log an error if the IO threw, and otherwise it will print the list of integers.

You should avoid unsafeRunSync for a number of reasons, discussed here. It's blocking, it can cause crashes, it can cause deadlocks, and it can halt your application.

If you really want to run your IO as a blocking computation, then you can precede this with attempt() to have your IO<A> become an IO<Either<Throwable, A>> similar to the unsafeRunAsync callback parameter. At least then you won't crash.

But unsafeRunAsync is preferred. Also, make sure your callback passed to unsafeRunAsync won't throw any errors, at it's assumed it won't. Docs.


问:
How to call Mongoose Find function to fetch all of the documents as an array?
I am still new to Node.js and Mongoose and have no idea why this code is not returning an array. Any suggestions?

function all(array){
    array = card.cards.find({}, (err, items) => {
        array = items
        return array
    })
}

let arr = []
all(arr)
console.log(arr)
```

The end result is just an empty array.

答:
You have 2 problems.

1. No way to reassign variable passed as parameter inside a function:

var a = 1;
function fn(b) {
  b = 2;
}
fn(a);
a === 1; // true
```

<p>Also it's a bad pattern to mutate data inside functions when it's not really needed. Better just return and assign in outer scope:</p>

<pre class="lang-js prettyprint-override"><code>function all() {
  return card.cards.find({});
}
all(a).then(console.log); // [...]
</code>```

But you can mutate an array without reassigning:

var a = [1, 2, 3];
function fn(array) {
  array.splice(0, Infinity, ...[5, 6, 7, 8]);
}
a; // [1, 2, 3]
fn(a);
a; // [5, 6, 7, 8]
```

<h3>2. Async calls to Mongoose and async-await/promises</h3>

<p>Still even if you return that from function <code>all</code> you have to wait the result.</p>

<p>Calls to mongoose methods are async. It means they return Promise objects (or promises) that implements interface with methods <code>then</code> and <code>catch</code>. The function you pass to <code>then</code> method will be called if promise will be resolved successfully, and the function in <code>catch</code> will be called if the promise will be rejected.</p>

<p>E.g.</p>

<pre class="lang-js prettyprint-override"><code>function p() {
  if (Math.random() > 0.5) {
    return Promise.resolve(1);
  } else {
    return Promise.reject(new Error('2'));
  }
}

p()
  .then(function (result) { /* will be called half of time for resolve branch */ })
  .catch(function (error) { /* will be called for reject branch */ })
</code>```

This code is similar to but uses syntactic sugar:

async function p() {
  if (Math.random() > 0.5) return 1; // Similar to return Promise.resolve(1);
  throw new Error('2'); // Similar to return Promise.reject(new Error('2'))
}
```

<p>To wait promises inside async function there is an <code>await</code> keyword.</p>

<pre class="lang-js prettyprint-override"><code>async function all() {
  try {
    return await cards.find({});

  } catch(error) {
    if (false /* somehow handle mongoose errors */) {
      // check error, if db not ready just retry or do sth else
      // …
    }

    // Just rethrow if we can't do any
    throw error;
  }
}
</code>```

Read more about async-await: https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Statements/async_function

Body:
I am still new to Node.js and Mongoose and have no idea why this code is not returning an array. Any suggestions?

function all(array){
    array = card.cards.find({}, (err, items) => {
        array = items
        return array
    })
}

let arr = []
all(arr)
console.log(arr)

The end result is just an empty array.

Body:
You have 2 problems.

1. No way to reassign variable passed as parameter inside a function:

var a = 1;
function fn(b) {
  b = 2;
}
fn(a);
a === 1; // true

Also it's a bad pattern to mutate data inside functions when it's not really needed. Better just return and assign in outer scope:

function all() {
  return card.cards.find({});
}
all(a).then(console.log); // [...]

But you can mutate an array without reassigning:

var a = [1, 2, 3];
function fn(array) {
  array.splice(0, Infinity, ...[5, 6, 7, 8]);
}
a; // [1, 2, 3]
fn(a);
a; // [5, 6, 7, 8]

2. Async calls to Mongoose and async-await/promises

Still even if you return that from function all you have to wait the result.

Calls to mongoose methods are async. It means they return Promise objects (or promises) that implements interface with methods then and catch. The function you pass to then method will be called if promise will be resolved successfully, and the function in catch will be called if the promise will be rejected.

E.g.

function p() {
  if (Math.random() > 0.5) {
    return Promise.resolve(1);
  } else {
    return Promise.reject(new Error('2'));
  }
}

p()
  .then(function (result) { /* will be called half of time for resolve branch */ })
  .catch(function (error) { /* will be called for reject branch */ })

This code is similar to but uses syntactic sugar:

async function p() {
  if (Math.random() > 0.5) return 1; // Similar to return Promise.resolve(1);
  throw new Error('2'); // Similar to return Promise.reject(new Error('2'))
}

To wait promises inside async function there is an await keyword.

async function all() {
  try {
    return await cards.find({});

  } catch(error) {
    if (false /* somehow handle mongoose errors */) {
      // check error, if db not ready just retry or do sth else
      // …
    }

    // Just rethrow if we can't do any
    throw error;
  }
}

Read more about async-await: https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Statements/async_function


问:
Divide a Set into Subsets of objects having equal field?
I am currently trying to implement a data structure in java, and would like to divide a set of input objects, into different subsets of objects having an equal field.

An example use case:

We want to divide a list of persons to sub list of persons being born a certain date.

Input: person1 born 1990, person2 born 2000, person3 born 1990.

Output:

1 -> person1, person3

2 -> person2

```<code>public Map<Integer, List<Foo>> getIntToFooMap(List<Foo> foos) {
    Map<Integer, List<Foo>> map = new TreeMap<>(); // need keys to be automatically ordered.
    List<Foo> foosWithSameSetId = new ArrayList<>();
    if (!foos.isEmpty) {
       for (Foo foo: foos) {
           for (Foo foo2: foos) {
               if (foo.getSetId().equals(foo2.getSetId())) {
                   foosWithSameSetId.add(foo2);
               }
           }
        map.put(foo.getSetId(), foosWithSameSetId);
        foosWithSameSetId.clear();
       }
    }
  return map;
 }
</code>```

The code above is not optimal, the time complexity is quadratic, and also not thread safe.
Can someone tell me of a better way to divide a List or a Set into subsets of objects having an equal field, in this case the setId.

答:
First, there's no need for the nested loop. You'd just get or create the set of the current foo's setId, and add foo to it:

```<code>for (Foo foo : foos) {
    map.computeIfAbsent(foo.getSetId(), i -> new ArrayList<>()).add(foo);
}
</code>```

Which would be equivalent to:

```<code>for (Foo foo : foos) {
    List<Foo> list = map.get(foo.getSetId());
    if(null == list) list = new ArrayList<>();
    list.add(foo);
}
</code>```

Now, you need to keep in mind the time complexity for your map implementation.

As an alternative, a groupingBy stream collector would add brevity to your code:

```<code>return foos.stream().collect(
        Collectors.groupingBy(Foo::getSetId, TreeMap::new, Collectors.toList()));
</code>```

Body:
I am currently trying to implement a data structure in java, and would like to divide a set of input objects, into different subsets of objects having an equal field.

An example use case:

We want to divide a list of persons to sub list of persons being born a certain date.

Input: person1 born 1990, person2 born 2000, person3 born 1990.

Output:

1 -> person1, person3

2 -> person2

public Map<Integer, List<Foo>> getIntToFooMap(List<Foo> foos) {
    Map<Integer, List<Foo>> map = new TreeMap<>(); // need keys to be automatically ordered.
    List<Foo> foosWithSameSetId = new ArrayList<>();
    if (!foos.isEmpty) {
       for (Foo foo: foos) {
           for (Foo foo2: foos) {
               if (foo.getSetId().equals(foo2.getSetId())) {
                   foosWithSameSetId.add(foo2);
               }
           }
        map.put(foo.getSetId(), foosWithSameSetId);
        foosWithSameSetId.clear();
       }
    }
  return map;
 }

The code above is not optimal, the time complexity is quadratic, and also not thread safe.
Can someone tell me of a better way to divide a List or a Set into subsets of objects having an equal field, in this case the setId.

Body:
First, there's no need for the nested loop. You'd just get or create the set of the current foo's setId, and add foo to it:

for (Foo foo : foos) {
    map.computeIfAbsent(foo.getSetId(), i -> new ArrayList<>()).add(foo);
}

Which would be equivalent to:

for (Foo foo : foos) {
    List<Foo> list = map.get(foo.getSetId());
    if(null == list) list = new ArrayList<>();
    list.add(foo);
}

Now, you need to keep in mind the time complexity for your map implementation.

As an alternative, a groupingBy stream collector would add brevity to your code:

return foos.stream().collect(
        Collectors.groupingBy(Foo::getSetId, TreeMap::new, Collectors.toList()));


问:
Inheritting type aliases in template classes
I want to do something like this:

My library defines a struct LibStruct

Clients of the library build their own structs that are somehow related (in the example, I used inheritance but maybe there's a better way of expressing)

The library has a method which does some operation using these structs

Avoid virtual function calls (hard requirement) and avoid union/variants (soft requirement)

What I've been doing so far is this:

```<code>template <typename T /** And a bunch more **/>
struct LibStruct {
  using Scalar = T;
  // And a bunch more aliases

  // Some data
};

template <typename T>
struct ClientStuctA : LibStruct<T> {
  using Scalar = T; // Can this be avoided?

  Scalar moreData;
};

struct ClientStuctB : LibStruct<double> {
  using Scalar = double; // Can this be avoided?

  Scalar otherData;
};

template <typename Whatever>
typename Whatever::Scalar doSomething(const Whatever& input) {
  // Do stuff
  return Whatever::Scalar();
}
</code>```

My issue with this is that all of the client structs need to redefine all the aliases so that doSomething can use them. Is there a way to avoid the need for that?

(Asking for C++14, but if there's a C++17 solution I'll take that too)

答:
So to sum up the back and forth me and @N.Shead had in the comments, supported by these links on why you can't do it better:

Propagating 'typedef' from based to derived class for 'template'

Inheritance and templates in C++ - why are inherited members invisible?

This seems to be the best we can do:

```<code>#include <iostream>

template<typename T /* And a bunch more */>
struct LibStruct {
  using Scalar = T;
  // ... and a lot more aliases
};

#define LIBSTRUCT_TYPES(base)           \
  using Scalar = typename base::Scalar; \
  // ... and a lot more

template<typename T>
struct ClientStuctA : LibStruct<T> {
  LIBSTRUCT_TYPES(LibStruct<T>)

  Scalar moreData;
  // OtherTypes moreDefinitions();

  Scalar func1() { return Scalar(2) * 3; }
  Scalar func2();
};

template<typename T>
typename ClientStuctA<T>::Scalar ClientStuctA<T>::func2() {
  return 5;
}

int main() {
  std::cout << ClientStuctA<int>().func1() << std::endl;
  std::cout << ClientStuctA<int>().func2() << std::endl;
  return 0;
}
</code>```

Body:
I want to do something like this:

My library defines a struct LibStruct

Clients of the library build their own structs that are somehow related (in the example, I used inheritance but maybe there's a better way of expressing)

The library has a method which does some operation using these structs

Avoid virtual function calls (hard requirement) and avoid union/variants (soft requirement)

What I've been doing so far is this:

template <typename T /** And a bunch more **/>
struct LibStruct {
  using Scalar = T;
  // And a bunch more aliases

  // Some data
};

template <typename T>
struct ClientStuctA : LibStruct<T> {
  using Scalar = T; // Can this be avoided?

  Scalar moreData;
};

struct ClientStuctB : LibStruct<double> {
  using Scalar = double; // Can this be avoided?

  Scalar otherData;
};

template <typename Whatever>
typename Whatever::Scalar doSomething(const Whatever& input) {
  // Do stuff
  return Whatever::Scalar();
}

My issue with this is that all of the client structs need to redefine all the aliases so that doSomething can use them. Is there a way to avoid the need for that?

(Asking for C++14, but if there's a C++17 solution I'll take that too)

Body:
So to sum up the back and forth me and @N.Shead had in the comments, supported by these links on why you can't do it better:

Propagating 'typedef' from based to derived class for 'template'

Inheritance and templates in C++ - why are inherited members invisible?

This seems to be the best we can do:

#include <iostream>

template<typename T /* And a bunch more */>
struct LibStruct {
  using Scalar = T;
  // ... and a lot more aliases
};

#define LIBSTRUCT_TYPES(base)           \
  using Scalar = typename base::Scalar; \
  // ... and a lot more

template<typename T>
struct ClientStuctA : LibStruct<T> {
  LIBSTRUCT_TYPES(LibStruct<T>)

  Scalar moreData;
  // OtherTypes moreDefinitions();

  Scalar func1() { return Scalar(2) * 3; }
  Scalar func2();
};

template<typename T>
typename ClientStuctA<T>::Scalar ClientStuctA<T>::func2() {
  return 5;
}

int main() {
  std::cout << ClientStuctA<int>().func1() << std::endl;
  std::cout << ClientStuctA<int>().func2() << std::endl;
  return 0;
}


问:
routing with model and variable
I am trying to test a custom email verification code with route model binding, when 2 wildcards are used, laravel always returns a 404.

this is my route in api.php

```<code>Route::get('/verify_contact_email/{id}/{hashed_key}', 'CustomEmailVerifyController@verifyContactEmail');
</code>```

this is the controller with verifyContactEmail

```<code>public function verifyContactEmail(UserContactEmailVerify $id, $hashed_key) { return $id; }
</code>```

when I remove the wildcard {hashed_key} and the $hashed_key, the model shows. I read up on laravel routing documentation, there is no mention of multiple wildcards or passing variable thru URL. Am I doing it wrong? Any help is much appreciated.

答:
You should try this</p>

```<code>public function verifyContactEmail($id, $hashed_key) {

   $id = UserContactEmailVerify::findOrFail($id);
   if(isset($id)){
       return $id;
   }

}
</code>```

Title: routing with model and variable

Body:
I am trying to test a custom email verification code with route model binding, when 2 wildcards are used, laravel always returns a 404.

this is my route in api.php

Route::get('/verify_contact_email/{id}/{hashed_key}', 'CustomEmailVerifyController@verifyContactEmail');

this is the controller with verifyContactEmail

public function verifyContactEmail(UserContactEmailVerify $id, $hashed_key) { return $id; }

when I remove the wildcard {hashed_key} and the $hashed_key, the model shows. I read up on laravel routing documentation, there is no mention of multiple wildcards or passing variable thru URL. Am I doing it wrong? Any help is much appreciated.

Body:
You should try this

public function verifyContactEmail($id, $hashed_key) {

   $id = UserContactEmailVerify::findOrFail($id);
   if(isset($id)){
       return $id;
   }

}


问:
Python list contains only digits integer or string
I need to check if a list contains only digits of int or str.

ex:

[3, 34, "45", 7, "3"]

would return true

and ["3" "hello", 9]

would return false

I know that the .isdigit() function cannot be used with a list containing int

答:
Simply try this one-linear pythonic solution:</p>

```<code>>>> all(map(lambda x: str(x).isdigit(), [3, 34, "45", 7, "3"]))
True
>>> all(map(lambda x: str(x).isdigit(), ["3" "hello", 9]))
False
</code>```

Body:
Simply try this one-linear pythonic solution:

>>> all(map(lambda x: str(x).isdigit(), [3, 34, "45", 7, "3"]))
True
>>> all(map(lambda x: str(x).isdigit(), ["3" "hello", 9]))
False


问:
umqtt.robust on Wemos
I am trying to install micropython-umqtt.robust on my Wemos D1 mini.
The way i tried this is as follow.

I use the Thonny editor

I have connected the wemos to the internet.

in wrepl type:

import upip

upip.install('micropython-umqtt.simple')

I get the folowing error: Installing to: /lib/
Error installing 'micropython-umqtt.simple': Package not found, packages may be partially        installed

upip.install('micropython-umqtt.robust')

I get the folowing error: Error installing 'micropython-umqtt.robust': Package not found,   packages may be partially installed

Can umqtt be installed on Wemos D1 mini ? if yes how do I do this ?

答:
I think the MicroPython build available from micropython.org already bundles MQTT so no need to install it with upip. Try this directly from the REPL:

from umqtt.robust import MQTTClient

or

from umqtt.simple import MQTTClient

and start using it from there

mqtt = MQTTClient(id, server, user, password)

Title: umqtt.robust on Wemos


问:
how to set and consume environment variables in node.js
so I have a  .env file in my TypeScript-node project. It contains a session variable like this:

```<code>SESSION_SECRET= EXAMPLEkeyHERE1
 BIRD_STRIPE_KEY= TheseAreNotReAlKeYS
CIRC_STRIPE_KEY= XXXXXX1232abc
</code>```

and Importing using :

import session = require('express-session'); const { SESSION_SECRET, PORT } = process.env;

and my app.js looks  like this:

```<code>if (!SESSION_SECRET) {
logger.error('No client secret. Set SESSION_SECRET environment variable.');
process.exit(1);
}

 /**
 * @description Create Express server.
 */
 const app = express();

if (IS_DEVELOPMENT) {
   /**
    * @description Error Handler. Provides full stack - remove for production
    */
 app.use(errorHandler());
 }

 /**
  * @description Express configuration
  */
  app.set('port', PORT || 3000);
  app.use(
   cors(),
   compression(),
   express.json(),
   session({
    secret: SESSION_SECRET,
    resave: true,
    saveUninitialized: false
})
);

/**
 * @description Express routers
 */
const routers = {
     api: express.Router(),
     provider: express.Router()
 };

 routers.api.use([routers.provider]);
 routers.provider.use('/provider', [routes.Bird, routes.Circ, routes.Lime, routes.Spin, routes.Tier]);

/**
 * @description These line will add /api/v{version} route prefix in every api requests.
 */
 app.use(`/api/v${API_CONFIG.versions.v1}`, routers.api);

/**
 * @description These lines are including error404 exported module and it will get called when routes not found.
 */
  app.use([middleware.Error404]);

 /**
 * @var server
 * @description Start Express server.
 */
 app.listen(app.get('port'), () => {
     console.log('App is running at http://localhost:%d in %s mode', app.get('port'), app.get('env'));
      console.log('Press CTRL-C to stop\n');
   });
</code>```

when I hit npm start I get this error:

```<code>> @ruler-mobility/ruler@1.0.0 serve /Users/macbook/Desktop/develop/rails-projects/ruler
> node src/app.js

 error: No client secret. Set SESSION_SECRET environment variable. {"service":"user-service"}
 npm ERR! code ELIFECYCLE
 npm ERR! errno 1
 npm ERR! @ruler-mobility/ruler@1.0.0 serve: `node src/app.js`
 npm ERR! Exit status 1
</code>```

so you can see the error is logged from the first if statement in the code. SESSION_SECRET is a falsy value and hence I'm getting  the error inside the if block

I have tried setting the SESSION_SECRET as a string SESSION_SECRET= "EXAMPLEkeyHERE1" in my .env file but it made no difference. can anyone tell me what I'm doing wrong here?

How do I access the variable from the .env file?

答:
You can use dotenv to load up env variables:

npm install dotenv -D
```

<p>and then as soon as possible in your script:</p>

<pre class="lang-js prettyprint-override"><code>require('dotenv').config()
</code>```

Here's the package

Body:
so I have a  .env file in my TypeScript-node project. It contains a session variable like this:

SESSION_SECRET= EXAMPLEkeyHERE1
 BIRD_STRIPE_KEY= TheseAreNotReAlKeYS
CIRC_STRIPE_KEY= XXXXXX1232abc

and Importing using :

import session = require('express-session'); const { SESSION_SECRET, PORT } = process.env;

and my app.js looks  like this:

if (!SESSION_SECRET) {
logger.error('No client secret. Set SESSION_SECRET environment variable.');
process.exit(1);
}

 /**
 * @description Create Express server.
 */
 const app = express();

if (IS_DEVELOPMENT) {
   /**
    * @description Error Handler. Provides full stack - remove for production
    */
 app.use(errorHandler());
 }

 /**
  * @description Express configuration
  */
  app.set('port', PORT || 3000);
  app.use(
   cors(),
   compression(),
   express.json(),
   session({
    secret: SESSION_SECRET,
    resave: true,
    saveUninitialized: false
})
);

/**
 * @description Express routers
 */
const routers = {
     api: express.Router(),
     provider: express.Router()
 };

 routers.api.use([routers.provider]);
 routers.provider.use('/provider', [routes.Bird, routes.Circ, routes.Lime, routes.Spin, routes.Tier]);

/**
 * @description These line will add /api/v{version} route prefix in every api requests.
 */
 app.use(`/api/v${API_CONFIG.versions.v1}`, routers.api);

/**
 * @description These lines are including error404 exported module and it will get called when routes not found.
 */
  app.use([middleware.Error404]);

 /**
 * @var server
 * @description Start Express server.
 */
 app.listen(app.get('port'), () => {
     console.log('App is running at http://localhost:%d in %s mode', app.get('port'), app.get('env'));
      console.log('Press CTRL-C to stop\n');
   });

when I hit npm start I get this error:

> @ruler-mobility/ruler@1.0.0 serve /Users/macbook/Desktop/develop/rails-projects/ruler
> node src/app.js

 error: No client secret. Set SESSION_SECRET environment variable. {"service":"user-service"}
 npm ERR! code ELIFECYCLE
 npm ERR! errno 1
 npm ERR! @ruler-mobility/ruler@1.0.0 serve: `node src/app.js`
 npm ERR! Exit status 1

so you can see the error is logged from the first if statement in the code. SESSION_SECRET is a falsy value and hence I'm getting  the error inside the if block

I have tried setting the SESSION_SECRET as a string SESSION_SECRET= "EXAMPLEkeyHERE1" in my .env file but it made no difference. can anyone tell me what I'm doing wrong here?

How do I access the variable from the .env file?

Body:
You can use dotenv to load up env variables:

npm install dotenv -D

and then as soon as possible in your script:

require('dotenv').config()

Here's the package