Provide a signalling table

The purpose of the signalling table is to provide a way how to request Debezium to execute a specific task and sychronize this request with the data coming from the database.
There are multiple scenarios where this feature can be useful

• snapshotting of tables upon user request
• refreshing a table schema out-of-bands - for example for PostgreSQL PK definition change
• resreshing s table schema without the need to monitor DDL or provide a DDL parser

The singalling table/collection will be defined in the connector config. If present it will be automatically whitelisted for capturing.
When the EventDispatcher receives message intended for signalling table then it will not deliver to Kafka topic (maybe it will for offset to commit offsets) but instead it will pass it to a CommandProcessor exposed by every connector.
There will be a DefaultCommandProcessor that will implement the behaviour provided by all connectors and every connector can extend it with its specific requirements.
It might be necessary to modify common framework code to provide an API to be called by the CommandProcessor.

The signalling table will contain three columns

• id - a unique identifier of the signal, probably UUID string if deduplication of execution is needed
• type - string representing type of the signal like refresh-schema, snapshot-tables, etc.
• parameters - s JSON string with parameters necessary for the given command

An example of snapshot-tables comand could be JSON like

{
  "tables": [
    {
      "tableId": "...",
      "snapshot": "data/schema",
      "query": "optional select to override default one"
    }
  ]
}

The singalling table will be implemented as a separate feature. In the implementation only a single command will be provided mostly as demo - log-message that will print message together with offsets into a log.

It is also possible that it will be necessary to extend offset context with a CommandContext. In this case addiitonal refactoring will be needed to introduce a set of shared contexts as we now use TransactionContext so connectors will not need to track multple contexts.

This feature can also be used for the implementation of watermark-based (chunked) snapshotting.