[
https://issues.apache.org/jira/browse/BEAM-4351?focusedWorklogId=104769&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-104769
]
ASF GitHub Bot logged work on BEAM-4351:
----------------------------------------
Author: ASF GitHub Bot
Created on: 22/May/18 19:49
Start Date: 22/May/18 19:49
Worklog Time Spent: 10m
Work Description: iemejia closed pull request #5451: [BEAM-4351] Enforce
ErrorProne analysis in mqtt IO
URL: https://github.com/apache/beam/pull/5451
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/sdks/java/io/mqtt/build.gradle b/sdks/java/io/mqtt/build.gradle
index d9eef843376..4db9f1b5366 100644
--- a/sdks/java/io/mqtt/build.gradle
+++ b/sdks/java/io/mqtt/build.gradle
@@ -17,13 +17,14 @@
*/
apply from: project(":").file("build_rules.gradle")
-applyJavaNature()
+applyJavaNature(failOnWarning: true)
description = "Apache Beam :: SDKs :: Java :: IO :: MQTT"
ext.summary = "IO to read and write to a MQTT broker."
dependencies {
compile library.java.guava
+ compileOnly library.java.findbugs_annotations
shadow project(path: ":beam-sdks-java-core", configuration: "shadow")
shadow library.java.slf4j_api
shadow library.java.joda_time
@@ -37,4 +38,5 @@ dependencies {
testCompile library.java.junit
testCompile library.java.hamcrest_core
testCompile library.java.slf4j_jdk14
+ testCompileOnly library.java.findbugs_annotations
}
diff --git
a/sdks/java/io/mqtt/src/main/java/org/apache/beam/sdk/io/mqtt/MqttIO.java
b/sdks/java/io/mqtt/src/main/java/org/apache/beam/sdk/io/mqtt/MqttIO.java
index c23fea68ffb..892bcf22d2b 100644
--- a/sdks/java/io/mqtt/src/main/java/org/apache/beam/sdk/io/mqtt/MqttIO.java
+++ b/sdks/java/io/mqtt/src/main/java/org/apache/beam/sdk/io/mqtt/MqttIO.java
@@ -23,6 +23,7 @@
import com.google.common.annotations.VisibleForTesting;
import java.io.IOException;
import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -543,7 +544,7 @@ public void createMqttClient() throws Exception {
@ProcessElement
public void processElement(ProcessContext context) throws Exception {
byte[] payload = context.element();
- LOG.debug("Sending message {}", new String(payload));
+ LOG.debug("Sending message {}", new String(payload,
StandardCharsets.UTF_8));
connection.publish(spec.connectionConfiguration().getTopic(), payload,
QoS.AT_LEAST_ONCE,
false);
}
diff --git
a/sdks/java/io/mqtt/src/test/java/org/apache/beam/sdk/io/mqtt/MqttIOTest.java
b/sdks/java/io/mqtt/src/test/java/org/apache/beam/sdk/io/mqtt/MqttIOTest.java
index 34a6a2f57f7..99c2cfedb5b 100644
---
a/sdks/java/io/mqtt/src/test/java/org/apache/beam/sdk/io/mqtt/MqttIOTest.java
+++
b/sdks/java/io/mqtt/src/test/java/org/apache/beam/sdk/io/mqtt/MqttIOTest.java
@@ -21,6 +21,7 @@
import static org.junit.Assert.assertTrue;
import java.net.ServerSocket;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Set;
import java.util.concurrent.ConcurrentSkipListSet;
@@ -85,16 +86,16 @@ public void testReadNoClientId() throws Exception {
.withMaxNumRecords(10);
PCollection<byte[]> output = pipeline.apply(mqttReader);
PAssert.that(output).containsInAnyOrder(
- "This is test 0".getBytes(),
- "This is test 1".getBytes(),
- "This is test 2".getBytes(),
- "This is test 3".getBytes(),
- "This is test 4".getBytes(),
- "This is test 5".getBytes(),
- "This is test 6".getBytes(),
- "This is test 7".getBytes(),
- "This is test 8".getBytes(),
- "This is test 9".getBytes()
+ "This is test 0".getBytes(StandardCharsets.UTF_8),
+ "This is test 1".getBytes(StandardCharsets.UTF_8),
+ "This is test 2".getBytes(StandardCharsets.UTF_8),
+ "This is test 3".getBytes(StandardCharsets.UTF_8),
+ "This is test 4".getBytes(StandardCharsets.UTF_8),
+ "This is test 5".getBytes(StandardCharsets.UTF_8),
+ "This is test 6".getBytes(StandardCharsets.UTF_8),
+ "This is test 7".getBytes(StandardCharsets.UTF_8),
+ "This is test 8".getBytes(StandardCharsets.UTF_8),
+ "This is test 9".getBytes(StandardCharsets.UTF_8)
);
// produce messages on the brokerService in another thread
@@ -117,8 +118,11 @@ public void testReadNoClientId() throws Exception {
}
}
for (int i = 0; i < 10; i++) {
- publishConnection.publish(topicName, ("This is test " +
i).getBytes(),
- QoS.EXACTLY_ONCE, false);
+ publishConnection.publish(
+ topicName,
+ ("This is test " + i).getBytes(StandardCharsets.UTF_8),
+ QoS.EXACTLY_ONCE,
+ false);
}
} catch (Exception e) {
// nothing to do
@@ -142,16 +146,16 @@ public void testRead() throws Exception {
"READ_PIPELINE"))
.withMaxReadTime(Duration.standardSeconds(3)));
PAssert.that(output).containsInAnyOrder(
- "This is test 0".getBytes(),
- "This is test 1".getBytes(),
- "This is test 2".getBytes(),
- "This is test 3".getBytes(),
- "This is test 4".getBytes(),
- "This is test 5".getBytes(),
- "This is test 6".getBytes(),
- "This is test 7".getBytes(),
- "This is test 8".getBytes(),
- "This is test 9".getBytes()
+ "This is test 0".getBytes(StandardCharsets.UTF_8),
+ "This is test 1".getBytes(StandardCharsets.UTF_8),
+ "This is test 2".getBytes(StandardCharsets.UTF_8),
+ "This is test 3".getBytes(StandardCharsets.UTF_8),
+ "This is test 4".getBytes(StandardCharsets.UTF_8),
+ "This is test 5".getBytes(StandardCharsets.UTF_8),
+ "This is test 6".getBytes(StandardCharsets.UTF_8),
+ "This is test 7".getBytes(StandardCharsets.UTF_8),
+ "This is test 8".getBytes(StandardCharsets.UTF_8),
+ "This is test 9".getBytes(StandardCharsets.UTF_8)
);
// produce messages on the brokerService in another thread
@@ -174,8 +178,11 @@ public void testRead() throws Exception {
}
}
for (int i = 0; i < 10; i++) {
- publishConnection.publish("READ_TOPIC", ("This is test " +
i).getBytes(),
- QoS.EXACTLY_ONCE, false);
+ publishConnection.publish(
+ "READ_TOPIC",
+ ("This is test " + i).getBytes(StandardCharsets.UTF_8),
+ QoS.EXACTLY_ONCE,
+ false);
}
} catch (Exception e) {
// nothing to do
@@ -219,7 +226,7 @@ public void testWrite() throws Exception {
try {
for (int i = 0; i < numberOfTestMessages; i++) {
Message message = connection.receive();
- messages.add(new String(message.getPayload()));
+ messages.add(new String(message.getPayload(),
StandardCharsets.UTF_8));
message.ack();
}
} catch (Exception e) {
@@ -230,7 +237,7 @@ public void testWrite() throws Exception {
ArrayList<byte[]> data = new ArrayList<>();
for (int i = 0; i < numberOfTestMessages; i++) {
- data.add(("Test " + i).getBytes());
+ data.add(("Test " + i).getBytes(StandardCharsets.UTF_8));
}
pipeline.apply(Create.of(data))
.apply(MqttIO.write()
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 104769)
Time Spent: 20m (was: 10m)
> Enforce ErrorProne analysis in the mqtt IO project
> --------------------------------------------------
>
> Key: BEAM-4351
> URL: https://issues.apache.org/jira/browse/BEAM-4351
> Project: Beam
> Issue Type: Improvement
> Components: io-java-mqtt
> Reporter: Scott Wegner
> Assignee: Tim Robertson
> Priority: Minor
> Labels: errorprone, starter
> Fix For: 2.5.0
>
> Time Spent: 20m
> Remaining Estimate: 0h
>
> Java ErrorProne static analysis was [recently
> enabled|https://github.com/apache/beam/pull/5161] in the Gradle build
> process, but only as warnings. ErrorProne errors are generally useful and
> easy to fix. Some work was done to [make sdks-java-core
> ErrorProne-clean|https://github.com/apache/beam/pull/5319] and add
> enforcement. This task is clean ErrorProne warnings and add enforcement in
> {{beam-sdks-java-io-mqtt}}. Additional context discussed on the [dev
> list|https://lists.apache.org/thread.html/95aae2785c3cd728c2d3378cbdff2a7ba19caffcd4faa2049d2e2f46@%3Cdev.beam.apache.org%3E].
> Fixing this issue will involve:
> # Follow instructions in the [Contribution
> Guide|https://beam.apache.org/contribute/] to set up a {{beam}} development
> environment.
> # Run the following command to compile and run ErrorProne analysis on the
> project: {{./gradlew :beam-sdks-java-io-mqtt:assemble}}
> # Fix each ErrorProne warning from the {{sdks/java/io/mqtt}} project.
> # In {{sdks/java/io/mqtt/build.gradle}}, add {{failOnWarning: true}} to the
> call the {{applyJavaNature()}}
> ([example|https://github.com/apache/beam/pull/5319/files#diff-9390c20635aed5f42f83b97506a87333R20]).
> This starter issue is sponsored by [~swegner]. Feel free to [reach
> out|https://beam.apache.org/community/contact-us/] with questions or code
> review:
> * JIRA: [~swegner]
> * GitHub: [@swegner|https://github.com/swegner]
> * Slack: [@Scott Wegner|https://s.apache.org/beam-slack-channel]
> * Email: swegner at google dot com
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
