tng/flink-connector-email

IMAP / SMTP connectors for Apache Flink®

flink-connector-email's Issues

Add useful functions

We could expose some hardcoded functions in the catalog, eg to find some header, address formatting, ...

Add a SMTP sink connector

Probably justifies renaming this repository and the project

Handle dropped connections and closed folders gracefully

Listen for removed messages

Flink 1.14

• Replace SubRowData
• Config enums implement DescribedEnum

Support forwarding arbitrary Jakarta properties

Option to mark ingested messages as seen

Provide table metadata

We should think about what metadata we might be able to provide, things coming to mind would be message count / unread message count.

Java 11

"Raw" metadata for SMTP to write emails more wholistically

Using new MimeMessage(session, InputStream). Other metadata would not be allowed in this case probably. Needs investigation.

Investigate why fetching content is slow

Potentially related: https://stackoverflow.com/questions/20237801/reading-from-javamail-takes-a-long-time

Start reading from a specified UID

Support HTML

Maybe we can do this through a content-type metadata and then use that?

IMAP Catalog

• Tables = Folders
• What are databases? Or can we just have only the default database and disallow create/alter/delete?

Support for updates on messages

Checkpointing for the IMAP connector

Some thoughts:

• We need to respect the UIDValidity to be aware of folders being deleted and re-created. See e.g. this.
• The synchronization block probably needs to be moved as we need to synchronize internal state updates, too.

Mapping of addresses is incorrect

When the list is empty (which can happen for e.g. CC, BCC, …), we always return an object rather than an empty list, regardless of how the column was specified. We should handle this case explicitly.

I haven't tested that this actually causes an IndexOutOfBounds, but it should.

Implement table statistics

Bounded mode

Add proper logging

Support write operations

• createTable
• dropTable
• renameTable

"Raw" metadata for IMAP source connector

Using Part#writeTo.

Switch to FLIP-27 source interfaces

Catalog connection can close

Exception in thread "main" org.apache.flink.table.client.SqlClientException: Unexpected exception. This is a bug. Please consider filing an issue.
	at org.apache.flink.table.client.SqlClient.main(SqlClient.java:213)
Caused by: org.apache.flink.table.catalog.exceptions.CatalogException: com.github.airblader.shaded.jakarta.mail.StoreClosedException: * BYE Jakarta Mail Exception: java.net.SocketException: Socket closed
	at com.github.airblader.imap.catalog.ImapCatalog.listTables(ImapCatalog.java:130)
	at org.apache.flink.table.catalog.CatalogManager.listTables(CatalogManager.java:447)
	at org.apache.flink.table.catalog.CatalogManager.listTables(CatalogManager.java:433)
	at org.apache.flink.table.api.internal.TableEnvironmentImpl.listTables(TableEnvironmentImpl.java:560)
	at org.apache.flink.table.client.gateway.local.LocalExecutor.lambda$listTables$5(LocalExecutor.java:345)
	at org.apache.flink.table.client.gateway.local.ExecutionContext.wrapClassLoader(ExecutionContext.java:255)
	at org.apache.flink.table.client.gateway.local.LocalExecutor.listTables(LocalExecutor.java:345)
	at org.apache.flink.table.client.cli.CliClient.callShowTables(CliClient.java:460)
	at org.apache.flink.table.client.cli.CliClient.callCommand(CliClient.java:289)
	at java.util.Optional.ifPresent(Optional.java:159)
	at org.apache.flink.table.client.cli.CliClient.open(CliClient.java:212)
	at org.apache.flink.table.client.SqlClient.openCli(SqlClient.java:142)
	at org.apache.flink.table.client.SqlClient.start(SqlClient.java:114)
	at org.apache.flink.table.client.SqlClient.main(SqlClient.java:201)
Caused by: com.github.airblader.shaded.jakarta.mail.StoreClosedException: * BYE Jakarta Mail Exception: java.net.SocketException: Socket closed
	at com.github.airblader.shaded.com.sun.mail.imap.IMAPFolder.throwClosedException(IMAPFolder.java:3708)
	at com.github.airblader.shaded.com.sun.mail.imap.IMAPFolder.doCommand(IMAPFolder.java:3842)
	at com.github.airblader.shaded.com.sun.mail.imap.DefaultFolder.list(DefaultFolder.java:56)
	at com.github.airblader.imap.catalog.ImapCatalog.listTables(ImapCatalog.java:118)

Support headers in SMTP sink connector

Evaluate how headers are returned

We currently return ArrayData of GenericRowData. Maybe there's a better way?

Add support for SupportsFilterPushDown