msgs = new ArrayList<>();
int messageCount = 0;
- // Resolve any in-flight transaction, committing unless there has been an error between
+ // Resolve any in-flight transaction, committing unless there has been an error
+ // between
// receiving the message from MQ and converting it
if (batchCompleteSignal != null) {
log.debug("Awaiting batch completion signal");
@@ -107,9 +111,10 @@ public MQSourceTask() {
reader.commit();
}
- // Increment the counter for the number of times poll is called so we can ensure we don't get stuck waiting for
+ // Increment the counter for the number of times poll is called so we can ensure
+ // we don't get stuck waiting for
// commitRecord callbacks to trigger the batch complete signal
- int currentPollCycle = pollCycle.incrementAndGet();
+ final int currentPollCycle = pollCycle.incrementAndGet();
log.debug("Starting poll cycle {}", currentPollCycle);
try {
@@ -123,64 +128,70 @@ public MQSourceTask() {
msgs.add(src);
messageCount++;
}
- } while ((src != null) && (messageCount < batchSize) && !stopNow.get());
- }
- else {
+ } while (src != null && messageCount < batchSize && !stopNow.get());
+ } else {
log.info("Stopping polling for records");
}
- }
- finally {
+ } finally {
}
- synchronized(this) {
+ synchronized (this) {
if (messageCount > 0) {
if (!stopNow.get()) {
batchCompleteSignal = new CountDownLatch(messageCount);
- }
- else {
- // Discard this batch - we've rolled back when the connection to MQ was closed in stop()
+ } else {
+ // Discard this batch - we've rolled back when the connection to MQ was closed
+ // in stop()
log.debug("Discarding a batch of {} records as task is stopping", messageCount);
msgs.clear();
batchCompleteSignal = null;
}
- }
- else {
+ } else {
batchCompleteSignal = null;
}
}
log.debug("Poll returning {} records", messageCount);
- log.trace("[{}] Exit {}.poll, retval={}", Thread.currentThread().getId(), this.getClass().getName(), messageCount);
+ log.trace("[{}] Exit {}.poll, retval={}", Thread.currentThread().getId(), this.getClass().getName(),
+ messageCount);
return msgs;
}
- /**
+ /**
*
- * Commit the offsets, up to the offsets that have been returned by {@link #poll()}. This
+ * Commit the offsets, up to the offsets that have been returned by
+ * {@link #poll()}. This
* method should block until the commit is complete.
*
*
- * SourceTasks are not required to implement this functionality; Kafka Connect will record offsets
- * automatically. This hook is provided for systems that also need to store offsets internally
+ * SourceTasks are not required to implement this functionality; Kafka Connect
+ * will record offsets
+ * automatically. This hook is provided for systems that also need to store
+ * offsets internally
* in their own system.
*
*/
public void commit() throws InterruptedException {
log.trace("[{}] Entry {}.commit", Thread.currentThread().getId(), this.getClass().getName());
- // This callback is simply used to ensure that the mechanism to use commitRecord callbacks
- // to check that all messages in a batch are complete is not getting stuck. If this callback
- // is being called, it means that Kafka Connect believes that all outstanding messages have
- // been completed. That should mean that commitRecord has been called for all of them too.
- // However, if too few calls to commitRecord are received, the connector could wait indefinitely.
- // If this commit callback is called twice without the poll cycle increasing, trigger the
+ // This callback is simply used to ensure that the mechanism to use commitRecord
+ // callbacks
+ // to check that all messages in a batch are complete is not getting stuck. If
+ // this callback
+ // is being called, it means that Kafka Connect believes that all outstanding
+ // messages have
+ // been completed. That should mean that commitRecord has been called for all of
+ // them too.
+ // However, if too few calls to commitRecord are received, the connector could
+ // wait indefinitely.
+ // If this commit callback is called twice without the poll cycle increasing,
+ // trigger the
// batch complete signal directly.
- int currentPollCycle = pollCycle.get();
+ final int currentPollCycle = pollCycle.get();
log.debug("Commit starting in poll cycle {}", currentPollCycle);
- if (lastCommitPollCycle == currentPollCycle)
- {
+ if (lastCommitPollCycle == currentPollCycle) {
synchronized (this) {
if (batchCompleteSignal != null) {
log.debug("Bumping batch complete signal by {}", batchCompleteSignal.getCount());
@@ -192,8 +203,7 @@ public void commit() throws InterruptedException {
}
}
}
- }
- else {
+ } else {
lastCommitPollCycle = currentPollCycle;
}
@@ -201,21 +211,28 @@ public void commit() throws InterruptedException {
}
/**
- * Signal this SourceTask to stop. In SourceTasks, this method only needs to signal to the task that it should stop
- * trying to poll for new data and interrupt any outstanding poll() requests. It is not required that the task has
- * fully stopped. Note that this method necessarily may be invoked from a different thread than {@link #poll()} and
+ * Signal this SourceTask to stop. In SourceTasks, this method only needs to
+ * signal to the task that it should stop
+ * trying to poll for new data and interrupt any outstanding poll() requests. It
+ * is not required that the task has
+ * fully stopped. Note that this method necessarily may be invoked from a
+ * different thread than {@link #poll()} and
* {@link #commit()}.
*
- * For example, if a task uses a {@link java.nio.channels.Selector} to receive data over the network, this method
- * could set a flag that will force {@link #poll()} to exit immediately and invoke
- * {@link java.nio.channels.Selector#wakeup() wakeup()} to interrupt any ongoing requests.
+ * For example, if a task uses a {@link java.nio.channels.Selector} to receive
+ * data over the network, this method
+ * could set a flag that will force {@link #poll()} to exit immediately and
+ * invoke
+ * {@link java.nio.channels.Selector#wakeup() wakeup()} to interrupt any ongoing
+ * requests.
*/
- @Override public void stop() {
+ @Override
+ public void stop() {
log.trace("[{}] Entry {}.stop", Thread.currentThread().getId(), this.getClass().getName());
stopNow.set(true);
- synchronized(this) {
+ synchronized (this) {
// Close the connection to MQ to clean up
if (reader != null) {
reader.close();
@@ -227,18 +244,23 @@ public void commit() throws InterruptedException {
/**
*
- * Commit an individual {@link SourceRecord} when the callback from the producer client is received, or if a record is filtered by a transformation.
+ * Commit an individual {@link SourceRecord} when the callback from the producer
+ * client is received, or if a record is filtered by a transformation.
*
*
- * SourceTasks are not required to implement this functionality; Kafka Connect will record offsets
- * automatically. This hook is provided for systems that also need to store offsets internally
+ * SourceTasks are not required to implement this functionality; Kafka Connect
+ * will record offsets
+ * automatically. This hook is provided for systems that also need to store
+ * offsets internally
* in their own system.
*
*
- * @param record {@link SourceRecord} that was successfully sent via the producer.
+ * @param record {@link SourceRecord} that was successfully sent via the
+ * producer.
* @throws InterruptedException
*/
- @Override public void commitRecord(SourceRecord record) throws InterruptedException {
+ @Override
+ public void commitRecord(final SourceRecord record) throws InterruptedException {
log.trace("[{}] Entry {}.commitRecord, record={}", Thread.currentThread().getId(), this.getClass().getName(), record);
synchronized (this) {
diff --git a/src/main/java/com/ibm/eventstreams/connect/mqsource/builders/BaseRecordBuilder.java b/src/main/java/com/ibm/eventstreams/connect/mqsource/builders/BaseRecordBuilder.java
index 399ddda..8a3c40e 100755
--- a/src/main/java/com/ibm/eventstreams/connect/mqsource/builders/BaseRecordBuilder.java
+++ b/src/main/java/com/ibm/eventstreams/connect/mqsource/builders/BaseRecordBuilder.java
@@ -37,12 +37,12 @@
public abstract class BaseRecordBuilder implements RecordBuilder {
private static final Logger log = LoggerFactory.getLogger(BaseRecordBuilder.class);
- public enum KeyHeader {NONE, MESSAGE_ID, CORRELATION_ID, CORRELATION_ID_AS_BYTES, DESTINATION};
+ public enum KeyHeader { NONE, MESSAGE_ID, CORRELATION_ID, CORRELATION_ID_AS_BYTES, DESTINATION };
protected KeyHeader keyheader = KeyHeader.NONE;
- private boolean copyJmsPropertiesFlag = Boolean.FALSE;
- private JmsToKafkaHeaderConverter jmsToKafkaHeaderConverter;
+ private boolean copyJmsPropertiesFlag = Boolean.FALSE;
+ private JmsToKafkaHeaderConverter jmsToKafkaHeaderConverter;
/**
* Configure this class.
@@ -51,55 +51,53 @@ public enum KeyHeader {NONE, MESSAGE_ID, CORRELATION_ID, CORRELATION_ID_AS_BYTES
*
* @throws ConnectException Operation failed and connector should stop.
*/
- @Override public void configure(Map props) {
- log.trace("[{}] Entry {}.configure, props={}", Thread.currentThread().getId(), this.getClass().getName(), props);
+ @Override public void configure(final Map props) {
+ log.trace("[{}] Entry {}.configure, props={}", Thread.currentThread().getId(), this.getClass().getName(),
+ props);
- String kh = props.get(MQSourceConnector.CONFIG_NAME_MQ_RECORD_BUILDER_KEY_HEADER);
+ final String kh = props.get(MQSourceConnector.CONFIG_NAME_MQ_RECORD_BUILDER_KEY_HEADER);
if (kh != null) {
if (kh.equals(MQSourceConnector.CONFIG_VALUE_MQ_RECORD_BUILDER_KEY_HEADER_JMSMESSAGEID)) {
keyheader = KeyHeader.MESSAGE_ID;
log.debug("Setting Kafka record key from JMSMessageID header field");
- }
- else if (kh.equals(MQSourceConnector.CONFIG_VALUE_MQ_RECORD_BUILDER_KEY_HEADER_JMSCORRELATIONID)) {
+ } else if (kh.equals(MQSourceConnector.CONFIG_VALUE_MQ_RECORD_BUILDER_KEY_HEADER_JMSCORRELATIONID)) {
keyheader = KeyHeader.CORRELATION_ID;
log.debug("Setting Kafka record key from JMSCorrelationID header field");
- }
- else if (kh.equals(MQSourceConnector.CONFIG_VALUE_MQ_RECORD_BUILDER_KEY_HEADER_JMSCORRELATIONIDASBYTES)) {
+ } else if (kh.equals(MQSourceConnector.CONFIG_VALUE_MQ_RECORD_BUILDER_KEY_HEADER_JMSCORRELATIONIDASBYTES)) {
keyheader = KeyHeader.CORRELATION_ID_AS_BYTES;
log.debug("Setting Kafka record key from JMSCorrelationIDAsBytes header field");
- }
- else if (kh.equals(MQSourceConnector.CONFIG_VALUE_MQ_RECORD_BUILDER_KEY_HEADER_JMSDESTINATION)) {
+ } else if (kh.equals(MQSourceConnector.CONFIG_VALUE_MQ_RECORD_BUILDER_KEY_HEADER_JMSDESTINATION)) {
keyheader = KeyHeader.DESTINATION;
log.debug("Setting Kafka record key from JMSDestination header field");
- }
- else {
+ } else {
log.error("Unsupported MQ record builder key header value {}", kh);
throw new ConnectException("Unsupported MQ record builder key header value");
}
}
- String str = props.get(MQSourceConnector.CONFIG_NAME_MQ_JMS_PROPERTY_COPY_TO_KAFKA_HEADER);
+ final String str = props.get(MQSourceConnector.CONFIG_NAME_MQ_JMS_PROPERTY_COPY_TO_KAFKA_HEADER);
copyJmsPropertiesFlag = Boolean.parseBoolean(Optional.ofNullable(str).orElse("false"));
jmsToKafkaHeaderConverter = new JmsToKafkaHeaderConverter();
- log.trace("[{}] Exit {}.configure", Thread.currentThread().getId(), this.getClass().getName());
- }
+ log.trace("[{}] Exit {}.configure", Thread.currentThread().getId(), this.getClass().getName());
+ }
/**
* Gets the key to use for the Kafka Connect SourceRecord.
*
- * @param context the JMS context to use for building messages
- * @param topic the Kafka topic
- * @param message the message
+ * @param context the JMS context to use for building messages
+ * @param topic the Kafka topic
+ * @param message the message
*
* @return the Kafka Connect SourceRecord's key
*
- * @throws JMSException Message could not be converted
+ * @throws JMSException Message could not be converted
*/
- public SchemaAndValue getKey(JMSContext context, String topic, Message message) throws JMSException {
+ public SchemaAndValue getKey(final JMSContext context, final String topic, final Message message)
+ throws JMSException {
Schema keySchema = null;
Object key = null;
- String keystr;
+ final String keystr;
switch (keyheader) {
case MESSAGE_ID:
@@ -107,8 +105,7 @@ public SchemaAndValue getKey(JMSContext context, String topic, Message message)
keystr = message.getJMSMessageID();
if (keystr.startsWith("ID:", 0)) {
key = keystr.substring(3);
- }
- else {
+ } else {
key = keystr;
}
break;
@@ -117,8 +114,7 @@ public SchemaAndValue getKey(JMSContext context, String topic, Message message)
keystr = message.getJMSCorrelationID();
if (keystr.startsWith("ID:", 0)) {
key = keystr.substring(3);
- }
- else {
+ } else {
key = keystr;
}
break;
@@ -140,36 +136,41 @@ public SchemaAndValue getKey(JMSContext context, String topic, Message message)
/**
* Gets the value to use for the Kafka Connect SourceRecord.
*
- * @param context the JMS context to use for building messages
- * @param topic the Kafka topic
- * @param messageBodyJms whether to interpret MQ messages as JMS messages
- * @param message the message
+ * @param context the JMS context to use for building messages
+ * @param topic the Kafka topic
+ * @param messageBodyJms whether to interpret MQ messages as JMS messages
+ * @param message the message
*
* @return the Kafka Connect SourceRecord's value
*
- * @throws JMSException Message could not be converted
+ * @throws JMSException Message could not be converted
*/
- public abstract SchemaAndValue getValue(JMSContext context, String topic, boolean messageBodyJms, Message message) throws JMSException;
+ public abstract SchemaAndValue getValue(JMSContext context, String topic, boolean messageBodyJms, Message message)
+ throws JMSException;
- /**
+ /**
* Convert a message into a Kafka Connect SourceRecord.
*
- * @param context the JMS context to use for building messages
- * @param topic the Kafka topic
- * @param messageBodyJms whether to interpret MQ messages as JMS messages
- * @param message the message
+ * @param context the JMS context to use for building messages
+ * @param topic the Kafka topic
+ * @param messageBodyJms whether to interpret MQ messages as JMS messages
+ * @param message the message
*
* @return the Kafka Connect SourceRecord
*
- * @throws JMSException Message could not be converted
+ * @throws JMSException Message could not be converted
*/
- @Override public SourceRecord toSourceRecord(JMSContext context, String topic, boolean messageBodyJms, Message message) throws JMSException {
- SchemaAndValue key = this.getKey(context, topic, message);
- SchemaAndValue value = this.getValue(context, topic, messageBodyJms, message);
-
- if (copyJmsPropertiesFlag && messageBodyJms)
- return new SourceRecord(null, null, topic, (Integer) null, key.schema(), key.value(), value.schema(), value.value(), message.getJMSTimestamp(), jmsToKafkaHeaderConverter.convertJmsPropertiesToKafkaHeaders(message));
+ @Override
+ public SourceRecord toSourceRecord(final JMSContext context, final String topic, final boolean messageBodyJms,
+ final Message message) throws JMSException {
+ final SchemaAndValue key = this.getKey(context, topic, message);
+ final SchemaAndValue value = this.getValue(context, topic, messageBodyJms, message);
+
+ if (copyJmsPropertiesFlag && messageBodyJms)
+ return new SourceRecord(null, null, topic, (Integer) null, key.schema(), key.value(), value.schema(),
+ value.value(), message.getJMSTimestamp(),
+ jmsToKafkaHeaderConverter.convertJmsPropertiesToKafkaHeaders(message));
else
- return new SourceRecord(null, null, topic, key.schema(), key.value(), value.schema(), value.value());
- }
+ return new SourceRecord(null, null, topic, key.schema(), key.value(), value.schema(), value.value());
+ }
}
\ No newline at end of file
diff --git a/src/main/java/com/ibm/eventstreams/connect/mqsource/builders/DefaultRecordBuilder.java b/src/main/java/com/ibm/eventstreams/connect/mqsource/builders/DefaultRecordBuilder.java
index 8d67a2b..acad1c5 100755
--- a/src/main/java/com/ibm/eventstreams/connect/mqsource/builders/DefaultRecordBuilder.java
+++ b/src/main/java/com/ibm/eventstreams/connect/mqsource/builders/DefaultRecordBuilder.java
@@ -55,28 +55,28 @@ public DefaultRecordBuilder() {
*
* @throws JMSException Message could not be converted
*/
- @Override public SchemaAndValue getValue(JMSContext context, String topic, boolean messageBodyJms, Message message) throws JMSException {
+ @Override public SchemaAndValue getValue(final JMSContext context, final String topic, final boolean messageBodyJms,
+ final Message message) throws JMSException {
Schema valueSchema = null;
Object value = null;
- // Interpreting the body as a JMS message type, we can accept BytesMessage and TextMessage only.
+ // Interpreting the body as a JMS message type, we can accept BytesMessage and
+ // TextMessage only.
// We do not know the schema so do not specify one.
if (messageBodyJms) {
if (message instanceof BytesMessage) {
log.debug("Bytes message with no schema");
value = message.getBody(byte[].class);
- }
- else if (message instanceof TextMessage) {
+ } else if (message instanceof TextMessage) {
log.debug("Text message with no schema");
value = message.getBody(String.class);
- }
- else {
+ } else {
log.error("Unsupported JMS message type {}", message.getClass());
throw new ConnectException("Unsupported JMS message type");
}
- }
- else {
- // Not interpreting the body as a JMS message type, all messages come through as BytesMessage.
+ } else {
+ // Not interpreting the body as a JMS message type, all messages come through as
+ // BytesMessage.
// In this case, we specify the value schema as OPTIONAL_BYTES.
log.debug("Bytes message with OPTIONAL_BYTES schema");
valueSchema = Schema.OPTIONAL_BYTES_SCHEMA;
diff --git a/src/main/java/com/ibm/eventstreams/connect/mqsource/builders/JsonRecordBuilder.java b/src/main/java/com/ibm/eventstreams/connect/mqsource/builders/JsonRecordBuilder.java
index 94f3927..ba3e389 100755
--- a/src/main/java/com/ibm/eventstreams/connect/mqsource/builders/JsonRecordBuilder.java
+++ b/src/main/java/com/ibm/eventstreams/connect/mqsource/builders/JsonRecordBuilder.java
@@ -15,7 +15,7 @@
*/
package com.ibm.eventstreams.connect.mqsource.builders;
-import static java.nio.charset.StandardCharsets.*;
+import static java.nio.charset.StandardCharsets.UTF_8;
import java.util.HashMap;
import javax.jms.BytesMessage;
@@ -45,7 +45,7 @@ public JsonRecordBuilder() {
converter = new JsonConverter();
// We just want the payload, not the schema in the output message
- HashMap m = new HashMap<>();
+ final HashMap m = new HashMap<>();
m.put("schemas.enable", "false");
// Convert the value, not the key (isKey == false)
@@ -55,26 +55,26 @@ public JsonRecordBuilder() {
/**
* Gets the value schema to use for the Kafka Connect SourceRecord.
*
- * @param context the JMS context to use for building messages
- * @param topic the Kafka topic
- * @param messageBodyJms whether to interpret MQ messages as JMS messages
- * @param message the message
+ * @param context the JMS context to use for building messages
+ * @param topic the Kafka topic
+ * @param messageBodyJms whether to interpret MQ messages as JMS messages
+ * @param message the message
*
* @return the Kafka Connect SourceRecord's value
*
- * @throws JMSException Message could not be converted
+ * @throws JMSException Message could not be converted
*/
- @Override public SchemaAndValue getValue(JMSContext context, String topic, boolean messageBodyJms, Message message) throws JMSException {
- byte[] payload;
+ @Override
+ public SchemaAndValue getValue(final JMSContext context, final String topic, final boolean messageBodyJms,
+ final Message message) throws JMSException {
+ final byte[] payload;
if (message instanceof BytesMessage) {
payload = message.getBody(byte[].class);
- }
- else if (message instanceof TextMessage) {
- String s = message.getBody(String.class);
+ } else if (message instanceof TextMessage) {
+ final String s = message.getBody(String.class);
payload = s.getBytes(UTF_8);
- }
- else {
+ } else {
log.error("Unsupported JMS message type {}", message.getClass());
throw new ConnectException("Unsupported JMS message type");
}
diff --git a/src/main/java/com/ibm/eventstreams/connect/mqsource/processor/JmsToKafkaHeaderConverter.java b/src/main/java/com/ibm/eventstreams/connect/mqsource/processor/JmsToKafkaHeaderConverter.java
index f7f9a19..52357ee 100644
--- a/src/main/java/com/ibm/eventstreams/connect/mqsource/processor/JmsToKafkaHeaderConverter.java
+++ b/src/main/java/com/ibm/eventstreams/connect/mqsource/processor/JmsToKafkaHeaderConverter.java
@@ -31,33 +31,33 @@
public class JmsToKafkaHeaderConverter {
private static final Logger log = LoggerFactory.getLogger(JmsToKafkaHeaderConverter.class);
- /**
- * Copies the JMS properties to Kafka headers.
- *
- * @param message JMS message.
- *
- * @return Kafka connect headers.
- */
- public ConnectHeaders convertJmsPropertiesToKafkaHeaders(Message message) {
- ConnectHeaders connectHeaders = new ConnectHeaders();
+ /**
+ * Copies the JMS properties to Kafka headers.
+ *
+ * @param message JMS message.
+ *
+ * @return Kafka connect headers.
+ */
+ public ConnectHeaders convertJmsPropertiesToKafkaHeaders(final Message message) {
+ final ConnectHeaders connectHeaders = new ConnectHeaders();
try {
@SuppressWarnings("unchecked")
- Enumeration propertyNames = (Enumeration)message.getPropertyNames();
- List jmsPropertyKeys = Collections.list(propertyNames);
+ final Enumeration propertyNames = (Enumeration) message.getPropertyNames();
+ final List jmsPropertyKeys = Collections.list(propertyNames);
jmsPropertyKeys.forEach(key -> {
try {
connectHeaders.addString(key.toString(), message.getObjectProperty(key.toString()).toString());
- }
- catch (JMSException e) {
- // Not failing the message processing if JMS properties cannot be read for some reason.
+ } catch (final JMSException e) {
+ // Not failing the message processing if JMS properties cannot be read for some
+ // reason.
log.warn("JMS exception {}", e);
}
});
- }
- catch (JMSException e) {
- // Not failing the message processing if JMS properties cannot be read for some reason.
+ } catch (final JMSException e) {
+ // Not failing the message processing if JMS properties cannot be read for some
+ // reason.
log.warn("JMS exception {}", e);
}
diff --git a/src/test/java/com/ibm/eventstreams/connect/mqsource/JmsToKafkaHeaderConverterTest.java b/src/test/java/com/ibm/eventstreams/connect/mqsource/JmsToKafkaHeaderConverterTest.java
index ed41bd1..0598a9b 100644
--- a/src/test/java/com/ibm/eventstreams/connect/mqsource/JmsToKafkaHeaderConverterTest.java
+++ b/src/test/java/com/ibm/eventstreams/connect/mqsource/JmsToKafkaHeaderConverterTest.java
@@ -44,17 +44,18 @@ public class JmsToKafkaHeaderConverterTest {
@Test
public void convertJmsPropertiesToKafkaHeaders() throws JMSException {
- List keys = Arrays.asList("facilityCountryCode", "facilityNum");
+ final List keys = Arrays.asList("facilityCountryCode", "facilityNum");
- Enumeration keyEnumeration = Collections.enumeration(keys);
+ final Enumeration keyEnumeration = Collections.enumeration(keys);
- //Arrange
+ // Arrange
when(message.getPropertyNames()).thenReturn(keyEnumeration);
when(message.getObjectProperty("facilityCountryCode")).thenReturn("US");
when(message.getObjectProperty("facilityNum")).thenReturn("12345");
- //Act
- ConnectHeaders actualConnectHeaders = jmsToKafkaHeaderConverter.convertJmsPropertiesToKafkaHeaders(message);
+ // Act
+ final ConnectHeaders actualConnectHeaders = jmsToKafkaHeaderConverter
+ .convertJmsPropertiesToKafkaHeaders(message);
//Verify
diff --git a/src/test/java/com/ibm/eventstreams/connect/mqsource/MQSourceConnectorTest.java b/src/test/java/com/ibm/eventstreams/connect/mqsource/MQSourceConnectorTest.java
index 3dd6b34..9ea8431 100644
--- a/src/test/java/com/ibm/eventstreams/connect/mqsource/MQSourceConnectorTest.java
+++ b/src/test/java/com/ibm/eventstreams/connect/mqsource/MQSourceConnectorTest.java
@@ -25,14 +25,14 @@
public class MQSourceConnectorTest {
@Test
public void testVersion() {
- String version = new MQSourceConnector().version();
- String expectedVersion = System.getProperty("connectorVersion");
+ final String version = new MQSourceConnector().version();
+ final String expectedVersion = System.getProperty("connectorVersion");
assertEquals("Expected connector version to match version of built jar file.", expectedVersion, version);
}
@Test
public void testConnectorType() {
- Connector connector = new MQSourceConnector();
+ final Connector connector = new MQSourceConnector();
assertTrue(SourceConnector.class.isAssignableFrom(connector.getClass()));
}
}
\ No newline at end of file