tng/flink-connector-email

IMAP / SMTP connectors for Apache Flink®

flink-connector-email's Issues

Implement table statistics

Add support for SupportsFilterPushDown

Support write operations

• createTable
• dropTable
• renameTable

Support headers in SMTP sink connector

Support HTML

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

Java 11

Listen for removed messages

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

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.

Add a SMTP sink connector

Probably justifies renaming this repository and the project

Support forwarding arbitrary Jakarta properties

Option to mark ingested messages as seen

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.

Bounded mode

Switch to FLIP-27 source interfaces

Handle dropped connections and closed folders gracefully

Flink 1.14

• Replace SubRowData
• Config enums implement DescribedEnum

"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.

Add proper logging

Support for updates on messages

IMAP Catalog

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

Add useful functions

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

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)

Evaluate how headers are returned

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

"Raw" metadata for IMAP source connector

Using Part#writeTo.

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.