openmessaging / openmessaging-java Goto Github PK

1、 The Consumer should support pulling the specified message through the queue, partition, and offset parameters.
2、Consumer bulk subscription queue. Such as: consumer.bindQueue (List queueNames);

1.In document OpenMessagingDomainArchitecture-v1.0.pdf Page 2, section Consumer, subsection PullConsumer. I believe that it may be "pull messages" rather than "poll messages" which is confusing.

• and use {@link ResourceManager#setConsumerProperties(String, KeyValue)} to modify.

ResourceManager#setConsumerProperties not exist

Hi!

Given the rising popularity of reactive APIs (also the Flow API in Java 9), such initiative as OpenMessaging should provide the first-class APIs for the reactive world.

To define the "reactive":

• Standard Publisher/Subscriber API - should be compatible with http://www.reactive-streams.org, so that many existing reactive solutions can easily integrate (RxJava, Akka, Reactor, etc...)
• Push/pull model - at least on end-user API level
• Back-pressure awareness - clients should be able to request their demand and producers should respect it
• [?] Smart resources utilization - ThreadPool-ing, async, non-blocking API

I pre-checked some of the points I think already provided. So far

The proven beauty of reactive APIs is that you can easily convert them to pull/push/streaming APIs, but not the other way around.
That might serve a foundation for the underlying implementations, but it's important to have it as an API, not the detail of implementation.

1. Modify return type to short type.
2. Add implementation.

I download the code, It's master branch, and the pom.xml's version is 1.0.0-beta-SNAPSHOT, then I checkout to branch 2.0.1-pubsub, I see the code is so different, even package!
So, is the branch 2.0.1-pubsub is RELEASED? Will the same big changes follow?

Offset and partition meta data should expose to the user.

For the intercepter APIs .

io.openmessaging.interceptor.ProducerInterceptor
io.openmessaging.interceptor.ConsumerInterceptor

It would be convenient for user to support java8 lambda style, that also make code more elegant.
Separating the preSend and the postSend method gives user more chorices.

Take a look the following example code:

Before:

        ProducerInterceptor interceptor = new ProducerInterceptor() {
            @Override
            public void preSend(Message message, Context attributes) {
                System.out.println("PreSend message: " + message);
            }

            @Override
            public void postSend(Message message, Context attributes) {
                System.out.println("PostSend message: " + message);
            }
        };
        producer.addInterceptor(interceptor);

After:

        producer.addPreSendInterceptor((message, attributes) ->
                System.out.println("PreSend message: " + message));

        producer.addPostSendInterceptor((message, attributes) ->
                System.out.println("PostSend message: " + message));

Besides Producer/Broker/Consumer topology, the Message/Stream communication use case could also apply to broker-less P2P communication. Basically, for the P2P communication, we could identify 4 communication styles:
Type 1. Oneway message/stream
Type 2. Request/Response
Type 3. Request/Stream
Type 4. Duplex stream
Currently the OpenMessaging API put pub/sub should be also able to support Type 1 P2P communication.
One possible way to enhance is to extend the interface SendResult

public interface SendResult {
    /**
     * The unique message id related to the {@code SendResult} instance.
     *
     * @return the message id.
     */
    String messageId();
}

For example, include a response in the RequestResponseSendResult, include a PeerTopic for the RequestStreamSendResult.

JD's messaging implementation supports sending multiple messages in a single transaction. Is there a need to improve the OpenMessaging API to be compatible with this feature?

Some attributes and interfaces are not universal in the current messaging and streaming field, but for now, most messaging middleware has implemented related concepts. such as store messages in different partitions, and in the form of the append-only log structure, assign each message with an offset for direct access.
So it is recommended to add extended attributes and interfaces to satisfy these message middleware. These attributes are not mandatory and may be removed or moved to mandatory fields.

Scenario: Suppose i already have a production instance of Kafka running in my infrastructure and i wanted to benchmark my Kafka cluster using this project.

Question: Does Openmessaging allow me to do that, without spinning a new AWS infra/Kafka cluster

If yes, can you please share an example to connect to external kafka endpoint ( any config file will be helpful as well.

direct message do not need dilivery into a queue, just direct send to a consumer

Prepare release OpenMessaging V0.2.0-alpha:

• Update the new domain arch
• Update to the target snapshot version
• Clear the related issues

Hi,

I was wondering if there were any reference implementation on an existing queue (Rabbit, Kafka, Pulsar...) ? Would it make sense to provide some in openmessaging-contrib or openmessaging-impl repo ? Maybe it's out of scope of this project.

Thanks

--
François

Hi guys,

I like a lot this initiative, but I am missing something common in advanced messaging schemas, some common meta headers and messaging patterns. Sorry if I just overlooked it while browsing in javadocs.

There exists already quite robust definitions of messaging patterns, especially for event-driven system or asynchronous communication there are some common message headers to be used like message Id, correlation Id, global transaction id, etc. I saw this framework only introduces generic key/value store like system and user headers within message.

Question: So this framework is not going to introduce some base messaging schema? It more focuses on high level fundamentals and leave the user to do what he wants? I think there could be added in time some kind of proposal for messaging patterns as low level implementation, what do you think?

We for example go to adopt IEC 61968-100 standard which target all possible situations on low level. Or OpenMessaging is more about foundation to be used to implement something like IEC messaging standard?

Thank you for answers.

Ladislav

Currently missing an authentication field, adding ACCOUNT_KEY for authentication.

As interface shows below：

    /**
     * <p>
     * Send batch messages to server.
     *
     * @param messages messages to be sent.
     */
    void send(List<Message> messages);

This API can lead to multiple behaviors, such as sending to multiple partitions, or the same partition, so how should we define the behavior of this method?

And this issue exists in many places, such as sequential messages, transactional message etc.
So I think we should polish this interface before we release the 1.0.0 version.

Copyright © 2017 OpenMessaging Project, a Linux Foundation Collaborative Project. All Rights Reserved unless explicitly granted under an open source license.

2017 --> 2018

Currently Future cannot be cancelled, adding cancel method to Future.

refer here, https://github.com/openmessaging/specification/blob/master/oms_access_point_schema.md

In order to achieve the ACK of messages, including single confirmation and batch confirmation, an interface needs to be added, so that vendors can better integrate with the existing products in the implementation, and is more flexible.

Currently SendResult's messageId is a String , in my case ,message id is a object with custom field ,i wonder if it can be a generic type like SendResult<T> ? Thanks for you time

public interface SendResult {
    /**
     * The unique message id related to the {@code SendResult} instance.
     *
     * @return the message id
     */
    String messageId();
}

The service package should be removed from OMS, while the first version of OMS only focus on Messaging and Streaming

java doc

• Refactor comments for each method

dynamically load the specific vendor implement

• URL component definition
• Load the implementation according to the URL
• Check if the specification version matches the implementation version

## Message filtering section
- [ ] support resource management api

Postponed to next release.

refactor properties

exception code & url

• Define OMS internal error code and provide a reference URL

Promise and Future

• Resolve issue #6

@vongosling @zhouxinyu @vintagewang If there are missed task, please add by yourself

In the traditional message queue, a consumer can only accept one model to receive a message, either push or pull, the current design is mainly used push model to receive messages, and the pull mode is used for special processing, which currently causes trouble to the user. Therefore, It would be better if we can split the two types consumer.

Message miss two important headers: RetryNumber and RetryReason.

• The RetryNumbers header field contains a number, which represents the number of message delivery.

• The RetryReason header field contains the text description of the reason that causes the last message delivery retry.

Hi everyone, as I had promised a lot time ago. I am going to participate the discussion about OpenMessaging. Before the discussion about details, I have some suggestions and questions.

• Is the OpenMessage a language independency specification?
  If not, ignore this. Otherwise, why the API descriptions are Java-specific? I think this is one of the most important thing at beginning. You said: this spec is vendor-neutral, you should be careful with that. Can't guarantee that, each vendor uses Java. Right?

• No specification documents can be found in the repository.
  I just found and read the pdf: https://github.com/openmessaging/openmessaging/blob/master/docs/OpenMessagingDomainArchitecture-v1.0.pdf. But from its name, it is not a specification, even not a alpha version. But the contents like one. Please correct me if I misunderstand.

• All documents should be in markdown format.
  Why markdown format is so important? The specification is for people to discuss and change, right? But, how can I change a pdf through a pr?

• No contributors or participators list.
  As you known, OpenTracing use OTSC and OTIAB, and also use github org to group people together( public members and badges in their profile). And I want to start a discuss, I know who should be mentioned in the issue. And be very helpful for spec populariztion. That is also Why I asked you to invite me into the group last time.

• Code should be isolated from this repository.
  Specification is nothing about Code, even in JSR, right? So even we are only Java-specific, we should move the codes to an independency repository.

@vongosling Hope these will help you.

Currently, there are method SendResult send(Message message) for sending messages synchronously and method Future<SendResult> sendAsync(Message message) for sending messages asynchronously.
Also, parameters retry number and acks number could be specified to perform at-most-once semantics.
In this way, I think we can remove the method sendOneway from Producer API.
How about this change?

1. Too many keyvalue params in the standard API
2. keyvalue is a weak datatype, but in messaging field, we usually use the base pre-defined properties and headers. some example like this:

new OMS.BaseProperties.Builder().deliveryMode().priority().build().
new OMS.BaseProperties.Builder().headers().build().

Sometimes we use OMS.builder().build(properties) to create the MessagingAccessPoint, it can work well in most scenarios.
But sometimes there are existed problem.

If we load the openmessaging-api in moduleA, and we use a customized ClassLoader to load the driver class, then we cannot get the driver class by Class.forName(driverImpl)

It is better to use Class.forName(String name, boolean initialize,ClassLoader loader) to get the driver class in line 50.

restore interface and implementation.

In the openmessaging-java api， MessageReceipt is mainly used in the following APIs

void ack(MessageReceipt receipt);
Message receive(String queueName, QueueMetaData queueMetaData, MessageReceipt messageReceipt, long timeout);
List<Message> batchReceive(String queueName, QueueMetaData queueMetaData, MessageReceipt messageReceipt,

But MessageReceipt is an empty interface，when used in ACK api, it should have get messageId API to confirmation message，when used in receive api, it should have getOffset API to pull message by offset，Otherwise, the API is not very useful and must be forced to convert to an implementation class, which obviously violates the Interface-oriented programming.

Is this a clerical error?

Background

Currently, Routing has an Operator pipeline, which is too complex to use, so remove it and use Expression is a better choice.

Design

Routing is composed of source, destination and expression:

• source and destination schema: "TOPIC::" or "QUEUE::"
• expression: "SQL::", or other expression types, even a http endpoint.

As we all know, JDK8 bring some cool features like lambda. And this is very convenient for users to process collections or streams. In the upcoming Stream API, it's better if we can support this.
So, I suggest to separate a jdk8 module/branch to add these features.
How about this idea?

Hope to provide more native ways to support Schema in PullConsumer in the future.

Improve message filtering APIs and support related resource management APIs

Currently KeyValue lacks a boolean type, adding boolean type to KeyValue.

A promise is a pledge to provide a result at some point in the future. See this document
io.openmessaging.Promise does not fit this semantic. It should be renamed to something like ListenableFuture following naming convention in existing literature and code base: Article, Guava ListenableFuture, JDK Listener, CPP std::future, Folly Future, Twitter future and Boost::future

Currently, OMS driver URL schema is:

oms:<driver_type>://<access_point>[,<access_point>,...]/<namespace>

The only user-oriented property is namespace, which is not enough to handle many complicated scenarios, like multiple tenants, environment isolation, AB tests, etc.

Consider adding region and account_id properties to OMS driver URL.

Currently, there are Producer API, Consumer API for basic IO operations on message queues.
For many scenes, users like to process messages in Streams.

The Stream API is lightweight, not as same as the distributed streaming systems like Apache Storm, Flink, Spark-Streaming. And its implementation can be embeded in any java applications.

It provides the most common stream operations like filter,flatMap,map,reduce,groupBy,join,count,max,min,window, etc.

It's different from the existing StreamingConsumer, which is a low level Consumer API for processing consumer positions, etc. (IMHO, the class name of StreamingConsumer should be renamed to a more reasonable one, the current might be confused for users.)