openmessaging/openconnect

OpenConnect proposes a specification for messaging, eventing and streaming connect scenario which provides various sources and sinks.

OpenConnect

Introduction

OpenConnect is a standard to connect between data sources and data destinations. Users could easily create connector instances with configurations via REST API.

There are two types of connectors: source connector and sink connector. A source connector is used for pulling data from a data source (e.g. RDBMS). The data is sent to corresponding message queue and expected to be consumed by one or many sink connectors. A sink connector receives message from the queue and loads into a data destination (e.g. data warehouse). Developers should implement source or sink connector interface to run their specific job.

Usually, connectors rely on a concrete message queue for data transportation. The message queue decouples source connectors from sink connectors. In the meantime, it provides capabilities such as failover, rate control and one to many data transportation etc. Some message queues (e.g. Kafka) provide bundled connect frameworks and a various of connectors developed officially or by the community. However, these frameworks are lack of interoperability, which means a connector developed for Kafka cannot run with RabbitMQ without modification and vice versa.

A connector follows OpenMessaging Connect could run with any message queues which support OpenMessaging API. OpenMessaging Connect provides a standalone runtime which uses OpenMessaging API for sending and consuming message, as well as the key/value operations for offset management.

Connector

void verifyAndSetConfig(KeyValue config)
Should invoke before start the connector.

void start()
Start the connector.

void stop()
Stop the connector.

void pause()
Pause the connector.

void resume()
Resume the connector.

Class<? extends Task> taskClass()
Returns the Task implementation for this Connector.

List<KeyValue> taskConfigs()
Returns a set of configurations for Task based on the current configuration.

Task

void start(KeyValue config)
Start the task with the given config.

void stop()
Stop the task.

void pause()
Pause the task.

void resume()
Resume the task.

Source task

Collection<SourceDataEntry> poll()
Return a collection of message entries to send.

Sink task

void put(Collection<SinkDataEntry> message)
Put the data entries to the sink.

License

openconnect's Issues

[New Feature]OpenMessaging Connect SQL Server

1. What is it？
   OpenMessaging connect platform is a framework that can be used to stream data to or from commonly used systems such as relational databases or MQ. Connector is the high level abstraction that coordinates data streaming by managing tasks. SQL Server connector is an implementation of OpenMessaging connector，by which stream data can be copy from one Sql Server instance to another, which can also realize the dynamic data capture.

2 How it could be?
TBD

You can refer to odar , which is an implementation of MySQL connector. docs

Connector lacks ConnectorContext

Connector lacks ConnectorContext's ability to actively reconfigure tasks

[New Feature] OpenMessaging Connect Oracle

1. What is it？
   OpenMessaging connect platform is a framework that can be used to stream data to or from commonly used systems such as relational databases or MQ. Connector is the high level abstraction that coordinates data streaming by managing tasks. Oracle connector is an implementation of OpenMessaging connector，by which stream data can be copy from one Oracle instance to another, which can also realize the dynamic data capture.

2 How it could be?
TBD

You can refer to odar , which is an implementation of MySQL connector. docs

[New Feature]OpenMessaging Connect Cassandra

1. What is it？
   OpenMessaging connect platform is a framework that can be used to stream data to or from commonly used systems such as databases or MQ. Connector is the high level abstraction that coordinates data streaming by managing tasks. Cassandra connector is an implementation of OpenMessaging connector，by which stream data can be copy from one Cassandra instance to another, which can also realize the dynamic data capture.

2 How it could be?
TBD

You can refer to odar , which is an implementation of MySQL connector. docs

License Headers Say Licensed to The ASF but this a Linux Foundation Project

License Headers are copied from the version for an Apache Software Foundation project, but OpenMessaging is a Linux Foundation project.

Header is:

    Licensed to the Apache Software Foundation (ASF) under one
    or more contributor license agreements.  See the NOTICE file
    distributed with this work for additional information
    regarding copyright ownership.  The ASF licenses this file
    to you under the Apache License, Version 2.0 (the
    "License"); you may not use this file except in compliance
    with the License.  You may obtain a copy of the License at

      http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing,
    software distributed under the License is distributed on an
    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
    KIND, either express or implied.  See the License for the
    specific language governing permissions and limitations
    under the License.

Header should look like this:

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at
 
    http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.

connect api upgrade

Connectrecord add a key field to identify the unique data

Connect record 增加字段来标识唯一

2019 Alibaba Summer of Code/2019 阿里巴巴编程之夏

2019 Alibaba Summer of Code

Welcome to the open source world! If you haven't planned how to spend this summer, come to the Alibaba Summer of Code and coding with us! 💻

Alibaba Summer of Code is a global program focused on engaging students directly into open source software development. Under the guidance of the mentor in the Alibaba open source project, students experience software development in the real world. Alibaba Summer of code will begin between June 28th and September 20th. Students can use the summertime to participate in the open source project and work with the core members of the project.

What you can get?

On this exclusive developer journey, students will have the opportunity to:

Participate in the top projects of the International Open Source Foundation;
Get a scholarship from Alibaba;
Obtain an open source contributor certificate;
Get a fast pass of Alibaba Internship/recruitment;
Get your code adopted and used by the open source project！

Our Mentor

duheng
github：duheng
Apache RocketMQ PMC member/Committer，Linux OpenMessaging TSC member

qingfeng
github：wlliqipeng
Apache RocketMQ China Community Initiator & Committer，Linux OpenMessaging Advisory Board Member

Apply Now!

You can click on this link to apply directly:
JDBC Connecotr
ElasticSearch Connector
Cassandra Connector
MongoDB Connector
full list of Alibaba Summer of Code Projects：
https://developer.aliyun.com/summerofcode2019

Timeline⏳

Contact us!

Email Address: [email protected]

The SchemaBuilder method conflicts with the Schema method

In io.openmessaging.connector.api.data.SchemaBuilder#build

     return new Schema(name, type, isOptional(), defaultValue, version, doc,
                fields == null ? null : Collections.unmodifiableList(new ArrayList<>(fields.values())), keySchema, valueSchema, 
  parameters);

In io.openmessaging.connector.api.data.Schema#addField

 public void addField(Field field) {
    this.fields.add(field);
    this.fieldsByName.put(field.getName(), field);
}

It creates an unmodifiable collection but Schema provides methods to add field,

Throws java.lang.UnsupportedOperationException: null if we call the add Field method

I wonder if we should not create unmodifiable collections

Optimize the source task commit method

Optimize the source task commit method

task context add configs

[New Feature] OpenMessaging Connect RabbitMQ

1. What is it？
   OpenMessaging connect platform is a framework that can be used to stream data to or from commonly used systems such as relational databases or MQ. Connector is the high level abstraction that coordinates data streaming by managing tasks. RabbitMQ connector is an implementation of OpenMessaging connector，by which stream data can be copy between RabbitMQ and RocketMQ.

2. How it could be?
   TBD

You can refer to odar , which is an implementation of MySQL connector. docs

[New Feature]OpenMessaging Connect Mangodb

1. What is it？
   OpenMessaging connect platform is a framework that can be used to stream data to or from commonly used systems such as databases or MQ. Connector is the high level abstraction that coordinates data streaming by managing tasks. S Mangodb connector is an implementation of OpenMessaging connector，by which stream data can be copy from one Mangodb instance to another, which can also realize the dynamic data capture.

2 How it could be?
TBD

You can refer to odar , which is an implementation of MySQL connector. docs

Optimize ConnectRecord api

问题如下：

1. 建议拆分 SinkRecord 和 SourceRecord， 语义明确，方便区分数据执行阶段，同时方便两部分api独立扩展，以防止出现有些字段在 source中用到，却在 sink中用不到但依然存在的情况，容易对使用上带来疑惑，造成过多的沟通的成本；
2. 显式声明topic / offset / queueId等字段，放到 recordPartition中用map 方式存储，结构较深，且对不熟悉代码的人不易发觉，学习成本较高，且导致代码过于松散
3. Record 绑定 topic, 方便多topic路由发送

Define error record reporting API

Add commitRecord support

add commitRecord method on SourceTask

Optimize API

1. Add struct object model
   统一struct构建方式，帮助快速完成struct的构建

2. Optimize empty method
   第一种：完全没必要实现的方法，需要隐掉
   第二种：可实现，可不实现的方法，需要设置default，不应该出现即可以在 A方法同时也可以在B 方法中实现的逻辑，比如 init 和 validate

3. Add sourcerecord and sinkrecord
   source 中的position和offset 在sink端中是不需要的，如果在sink端消费record时，对api中字段的释义不明确，就会造成很多的社区沟通成本，这样也会显的api不是很专一，所以抽象的好处是能让开发者看到的字段和api都是能可解释的且都是有明确含义的；这样也会对未来sink和source对象的独立的功能扩展和开发带来便利

4. Add offset writer abstraction
   包装RecordOffset使用使用，保证每个任务的offset的唯一性

..........

[New Feature] RocketMQ

Add connect exception mechanism

Connect API should have its own exception mechanism, not only to declare every exception but also can be used for the interaction with connect runtime.

[New Feature]OpenMessaging Connect Elasticsearch.

What is it？
OpenMessaging connect platform is a framework that can be used to stream data to or from commonly used systems such as elasticsearch or relational databases or MQ. Connector is the high level abstraction that coordinates data streaming by managing tasks. Elasticsearch connector is an implementation of OpenMessaging connector，by which stream data can be copy from MQ to elasticsearch.
2 How it could be?
TBD

You can refer to odar , which is an implementation of MySQL connector. docs

OpenMessaging Connect JDBC

1. What is it？
   OpenMessaging connect platform is a framework that can be used to stream data to or from commonly used systems such as relational databases or MQ. Connector is the high level abstraction that coordinates data streaming by managing tasks.
   JDBC connector, including source connector and slink connector, allow you to exchange data between relational databases and rocketmq. The JDBC source connector allows you to import data from any relational database with a JDBC driver into rocketmq topics.The JDBC sink connector allows you to export data from rocketmq topics to any relational database with a JDBC driver.

2 How it could be?
TBD

You can refer to odar , which is an implementation of MySQL connector. docs

[Typo] typo in DataEntryBuilder.java, getFiled() should getField()

I found a place that is likely a typo. But I found it is used widely in downstream projects, fixing this requires a typo fix tool and some planning. Should we consider fixing it now or put it for a later refactoring?

// wondering putField() would make more sense
    public DataEntryBuilder putFiled(String fieldName, Object value) {

        Field field = lookupField(fieldName);
        payload[field.getIndex()] = value;
        return this;
    }

Add Sharding mechanism support

Sharding is a topic that cannot be bypassed for most messaging middleware, so we need to add sharding support in OpenMessaging connect.

Optimize transform api

[New Feature] OpenMessaging Connect Kafka

1. What is it？
   OpenMessaging connect platform is a framework that can be used to stream data to or from commonly used systems such as relational databases or MQ. Connector is the high level abstraction that coordinates data streaming by managing tasks. kafka connector is an implementation of OpenMessaging connector，by which stream data can be copy between RabbitMQ and RocketMQ.

2. How it could be?
   TBD

You can refer to odar , which is an implementation of MySQL connector. docs

Potentially NPE

In io.openmessaging.connector.api.data.Schema#addField

If fields are not passed through the constructor, calling the method directly returns a NPE