New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[BEAM-13] Add JmsIO #299
[BEAM-13] Add JmsIO #299
Changes from all commits
604bb48
ffe1498
7bdec54
ecf6acf
b608f19
a65a858
d73cc4d
1f028d8
18798ab
3ec9839
bd08171
7c748f6
2c8999c
93ccc62
de4522a
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,136 @@ | ||
<?xml version="1.0" encoding="UTF-8"?> | ||
<!-- | ||
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. | ||
--> | ||
<project xmlns="http://maven.apache.org/POM/4.0.0" | ||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" | ||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> | ||
|
||
<modelVersion>4.0.0</modelVersion> | ||
|
||
<parent> | ||
<groupId>org.apache.beam</groupId> | ||
<artifactId>beam-sdks-java-io-parent</artifactId> | ||
<version>0.2.0-incubating-SNAPSHOT</version> | ||
<relativePath>../pom.xml</relativePath> | ||
</parent> | ||
|
||
<artifactId>jms</artifactId> | ||
<name>Apache Beam :: SDKs :: Java :: IO :: JMS</name> | ||
<description>IO to read and write to JMS (Java Messaging Service) | ||
destinations (queues and topics). </description> | ||
|
||
<properties> | ||
<activemq.version>5.13.1</activemq.version> | ||
<geronimo-jms.version>1.1.1</geronimo-jms.version> | ||
</properties> | ||
|
||
<build> | ||
<plugins> | ||
<plugin> | ||
<groupId>org.apache.maven.plugins</groupId> | ||
<artifactId>maven-compiler-plugin</artifactId> | ||
</plugin> | ||
<plugin> | ||
<groupId>org.apache.maven.plugins</groupId> | ||
<artifactId>maven-jar-plugin</artifactId> | ||
</plugin> | ||
<plugin> | ||
<groupId>org.apache.maven.plugins</groupId> | ||
<artifactId>maven-source-plugin</artifactId> | ||
</plugin> | ||
<plugin> | ||
<groupId>org.apache.maven.plugins</groupId> | ||
<artifactId>maven-checkstyle-plugin</artifactId> | ||
</plugin> | ||
</plugins> | ||
</build> | ||
|
||
<dependencies> | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We should analyze the new dependencies, for license-compliance, size, stability, etc. Can you make sure we log this in the review by pasting the output of the verbose dependency tree and licenses. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Good idea, let me prepare this. Maybe we can add a note in the github template about that (for new PR). There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. JB: still todo? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I checked the dependencies license and it's all Apache (as we are using Apache projects ;)). |
||
<dependency> | ||
<groupId>org.apache.beam</groupId> | ||
<artifactId>beam-sdks-java-core</artifactId> | ||
</dependency> | ||
|
||
<dependency> | ||
<groupId>org.slf4j</groupId> | ||
<artifactId>slf4j-api</artifactId> | ||
</dependency> | ||
|
||
<dependency> | ||
<groupId>joda-time</groupId> | ||
<artifactId>joda-time</artifactId> | ||
</dependency> | ||
|
||
<dependency> | ||
<groupId>com.google.guava</groupId> | ||
<artifactId>guava</artifactId> | ||
</dependency> | ||
|
||
<dependency> | ||
<groupId>org.apache.geronimo.specs</groupId> | ||
<artifactId>geronimo-jms_1.1_spec</artifactId> | ||
<version>${geronimo-jms.version}</version> | ||
</dependency> | ||
|
||
<dependency> | ||
<groupId>com.google.code.findbugs</groupId> | ||
<artifactId>annotations</artifactId> | ||
</dependency> | ||
|
||
<!-- test dependencies --> | ||
<dependency> | ||
<groupId>org.apache.activemq</groupId> | ||
<artifactId>activemq-broker</artifactId> | ||
<version>${activemq.version}</version> | ||
<scope>test</scope> | ||
</dependency> | ||
<dependency> | ||
<groupId>org.apache.activemq</groupId> | ||
<artifactId>activemq-kahadb-store</artifactId> | ||
<version>${activemq.version}</version> | ||
<scope>test</scope> | ||
</dependency> | ||
<dependency> | ||
<groupId>org.apache.activemq</groupId> | ||
<artifactId>activemq-client</artifactId> | ||
<version>${activemq.version}</version> | ||
<scope>test</scope> | ||
</dependency> | ||
<dependency> | ||
<groupId>org.apache.beam</groupId> | ||
<artifactId>beam-runners-direct-java</artifactId> | ||
<version>${project.version}</version> | ||
<scope>test</scope> | ||
</dependency> | ||
<dependency> | ||
<groupId>junit</groupId> | ||
<artifactId>junit</artifactId> | ||
<scope>test</scope> | ||
</dependency> | ||
<dependency> | ||
<groupId>org.hamcrest</groupId> | ||
<artifactId>hamcrest-all</artifactId> | ||
<scope>test</scope> | ||
</dependency> | ||
<dependency> | ||
<groupId>org.slf4j</groupId> | ||
<artifactId>slf4j-jdk14</artifactId> | ||
<scope>test</scope> | ||
</dependency> | ||
</dependencies> | ||
|
||
</project> |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,82 @@ | ||
/* | ||
* 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. | ||
*/ | ||
package org.apache.beam.sdk.io.jms; | ||
|
||
import org.apache.beam.sdk.coders.AvroCoder; | ||
import org.apache.beam.sdk.coders.DefaultCoder; | ||
import org.apache.beam.sdk.io.UnboundedSource; | ||
import org.apache.beam.sdk.transforms.windowing.BoundedWindow; | ||
|
||
import org.joda.time.Instant; | ||
|
||
import java.util.ArrayList; | ||
import java.util.List; | ||
|
||
import javax.jms.Message; | ||
|
||
/** | ||
* Checkpoint for an unbounded JmsIO.Read. Consists of | ||
* JMS destination name, and the latest message ID consumed so far. | ||
*/ | ||
@DefaultCoder(AvroCoder.class) | ||
public class JmsCheckpointMark implements UnboundedSource.CheckpointMark { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Do take a quick pass over this class and try to restrict the visible APIs as much as possible. In general, anything that's an internal implementation detail should be There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. +1. Fixing that. |
||
|
||
private final List<Message> messages = new ArrayList<>(); | ||
private Instant oldestPendingTimestamp = BoundedWindow.TIMESTAMP_MIN_VALUE; | ||
|
||
public JmsCheckpointMark() { | ||
} | ||
|
||
protected List<Message> getMessages() { | ||
return this.messages; | ||
} | ||
|
||
protected void addMessage(Message message) throws Exception { | ||
Instant currentMessageTimestamp = new Instant(message.getJMSTimestamp()); | ||
if (currentMessageTimestamp.isBefore(oldestPendingTimestamp)) { | ||
oldestPendingTimestamp = currentMessageTimestamp; | ||
} | ||
messages.add(message); | ||
} | ||
|
||
protected Instant getOldestPendingTimestamp() { | ||
return oldestPendingTimestamp; | ||
} | ||
|
||
/** | ||
* Acknowledge all outstanding message. Since we believe that messages will be delivered in | ||
* timestamp order, and acknowledged messages will not be retried, the newest message in this | ||
* batch is a good bound for future messages. | ||
*/ | ||
@Override | ||
public void finalizeCheckpoint() { | ||
for (Message message : messages) { | ||
try { | ||
message.acknowledge(); | ||
Instant currentMessageTimestamp = new Instant(message.getJMSTimestamp()); | ||
if (currentMessageTimestamp.isAfter(oldestPendingTimestamp)) { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Add a comment here, something like:
(Could also use oldest message for a more conservative bound in case some messages do get retried anyway, but that's up to you.) There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Good idea. |
||
oldestPendingTimestamp = currentMessageTimestamp; | ||
} | ||
} catch (Exception e) { | ||
// nothing to do | ||
} | ||
} | ||
messages.clear(); | ||
} | ||
|
||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
beam-sdks-java-io-jms
?