Solace Read connector

sdks/java/io/solace/build.gradle

@@ -35,8 +35,16 @@ dependencies {
     implementation library.java.slf4j_api
     implementation library.java.joda_time
     implementation library.java.solace
+    implementation library.java.vendored_guava_32_1_2_jre
     implementation project(":sdks:java:extensions:avro")
     implementation library.java.vendored_grpc_1_60_1
     implementation library.java.avro
     permitUnusedDeclared library.java.avro
+    implementation library.java.vendored_grpc_1_60_1
+
+    testImplementation library.java.junit
+    testImplementation project(path: ":sdks:java:io:common", configuration: "testRuntimeMigration")
+    testImplementation project(path: ":sdks:java:testing:test-utils", configuration: "testRuntimeMigration")
+    testRuntimeOnly library.java.slf4j_jdk14
+    testRuntimeOnly project(path: ":runners:direct-java", configuration: "shadow")
 }

sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/broker/SessionServiceFactory.java

@@ -17,17 +17,36 @@
  */
 package org.apache.beam.sdk.io.solace.broker;
 
+import com.solacesystems.jcsmp.Queue;
 import java.io.Serializable;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 /**
  * This abstract class serves as a blueprint for creating `SessionService` objects. It introduces a
  * queue property and mandates the implementation of a create() method in concrete subclasses.
  */
 public abstract class SessionServiceFactory implements Serializable {
 
+  /**
+   * A reference to a Queue object. This is set when the pipline is constructed (in the {@link
+   * org.apache.beam.sdk.io.solace.SolaceIO.Read#expand(org.apache.beam.sdk.values.PBegin)} method).
+   * This could be used to associate the created SessionService with a specific queue for message
+   * handling.
+   */
+  @Nullable Queue queue;
+
   /**
    * This is the core method that subclasses must implement. It defines how to construct and return
    * a SessionService object.
    */
   public abstract SessionService create();
+
+  /**
+   * This method is called in the {@link
+   * org.apache.beam.sdk.io.solace.SolaceIO.Read#expand(org.apache.beam.sdk.values.PBegin)} method
+   * to set the Queue reference.
+   */
+  public void setQueue(Queue queue) {
+    this.queue = queue;
+  }
 }

sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/data/Solace.java

@@ -21,6 +21,8 @@
 import com.solacesystems.jcsmp.BytesXMLMessage;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import org.apache.beam.sdk.schemas.AutoValueSchema;
+import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
 import org.apache.beam.vendor.grpc.v1p60p1.com.google.protobuf.ByteString;
 import org.checkerframework.checker.nullness.qual.Nullable;
 import org.slf4j.Logger;
@@ -74,6 +76,7 @@ public enum DestinationType {
 
   /** Represents a Solace message destination (either a Topic or a Queue). */
   @AutoValue
+  @DefaultSchema(AutoValueSchema.class)
   public abstract static class Destination {
     /**
      * Gets the name of the destination.
@@ -105,6 +108,7 @@ abstract static class Builder {
 
   /** Represents a Solace message record with its associated metadata. */
   @AutoValue
+  @DefaultSchema(AutoValueSchema.class)
   public abstract static class Record {
     /**
      * Gets the unique identifier of the message, a string for an application-specific message

sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/MockEmptySessionService.java

@@ -0,0 +1,46 @@
+/*
+ * 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.solace;
+
+import org.apache.beam.sdk.io.solace.broker.MessageReceiver;
+import org.apache.beam.sdk.io.solace.broker.SessionService;
+
+public class MockEmptySessionService implements SessionService {
+
+  String exceptionMessage = "This is an empty client, use a MockSessionService instead.";
+
+  @Override
+  public void close() {
+    throw new UnsupportedOperationException(exceptionMessage);
+  }
+
+  @Override
+  public boolean isClosed() {
+    throw new UnsupportedOperationException(exceptionMessage);
+  }
+
+  @Override
+  public MessageReceiver createReceiver() {
+    throw new UnsupportedOperationException(exceptionMessage);
+  }
+
+  @Override
+  public void connect() {
+    throw new UnsupportedOperationException(exceptionMessage);
+  }
+}

sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/MockSempClient.java

@@ -0,0 +1,87 @@
+/*
+ * 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.solace;
+
+import com.solacesystems.jcsmp.JCSMPFactory;
+import com.solacesystems.jcsmp.Queue;
+import java.io.IOException;
+import org.apache.beam.sdk.io.solace.broker.SempClient;
+import org.apache.beam.sdk.transforms.SerializableFunction;
+
+public class MockSempClient implements SempClient {
+
+  private final SerializableFunction<String, Boolean> isQueueNonExclusiveFn;
+  private final SerializableFunction<String, Long> getBacklogBytesFn;
+  private final SerializableFunction<String, Integer> createQueueForTopicFn;
+
+  private MockSempClient(
+      SerializableFunction<String, Boolean> isQueueNonExclusiveFn,
+      SerializableFunction<String, Long> getBacklogBytesFn,
+      SerializableFunction<String, Integer> createQueueForTopicFn) {
+    this.isQueueNonExclusiveFn = isQueueNonExclusiveFn;
+    this.getBacklogBytesFn = getBacklogBytesFn;
+    this.createQueueForTopicFn = createQueueForTopicFn;
+  }
+
+  public static Builder builder() {
+    return new Builder();
+  }
+
+  public static class Builder {
+    private SerializableFunction<String, Boolean> isQueueNonExclusiveFn = (queueName) -> true;
+    private SerializableFunction<String, Long> getBacklogBytesFn = (queueName) -> 0L;
+    private SerializableFunction<String, Integer> createQueueForTopicFn = (queueName) -> 0;
+
+    public Builder setIsQueueNonExclusiveFn(
+        SerializableFunction<String, Boolean> isQueueNonExclusiveFn) {
+      this.isQueueNonExclusiveFn = isQueueNonExclusiveFn;
+      return this;
+    }
+
+    public Builder setGetBacklogBytesFn(SerializableFunction<String, Long> getBacklogBytesFn) {
+      this.getBacklogBytesFn = getBacklogBytesFn;
+      return this;
+    }
+
+    public Builder setCreateQueueForTopicFn(
+        SerializableFunction<String, Integer> createQueueForTopicFn) {
+      this.createQueueForTopicFn = createQueueForTopicFn;
+      return this;
+    }
+
+    public MockSempClient build() {
+      return new MockSempClient(isQueueNonExclusiveFn, getBacklogBytesFn, createQueueForTopicFn);
+    }
+  }
+
+  @Override
+  public boolean isQueueNonExclusive(String queueName) throws IOException {
+    return isQueueNonExclusiveFn.apply(queueName);
+  }
+
+  @Override
+  public Queue createQueueForTopic(String queueName, String topicName) throws IOException {
+    createQueueForTopicFn.apply(queueName);
+    return JCSMPFactory.onlyInstance().createQueue(queueName);
+  }
+
+  @Override
+  public long getBacklogBytes(String queueName) throws IOException {
+    return getBacklogBytesFn.apply(queueName);
+  }
+}

sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/MockSempClientFactory.java

@@ -0,0 +1,38 @@
+/*
+ * 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.solace;
+
+import org.apache.beam.sdk.io.solace.broker.SempClient;
+import org.apache.beam.sdk.io.solace.broker.SempClientFactory;
+
+public class MockSempClientFactory implements SempClientFactory {
+  SempClient sempClient;
+
+  public MockSempClientFactory(SempClient sempClient) {
+    this.sempClient = sempClient;
+  }
+
+  public static SempClientFactory getDefaultMock() {
+    return new MockSempClientFactory(MockSempClient.builder().build());
+  }
+
+  @Override
+  public SempClient create() {
+    return sempClient;
+  }
+}

sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/MockSessionService.java

@@ -0,0 +1,88 @@
+/*
+ * 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.solace;
+
+import com.solacesystems.jcsmp.BytesXMLMessage;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.beam.sdk.io.solace.broker.MessageReceiver;
+import org.apache.beam.sdk.io.solace.broker.SessionService;
+import org.apache.beam.sdk.transforms.SerializableFunction;
+
+public class MockSessionService implements SessionService {
+
+  private final SerializableFunction<Integer, BytesXMLMessage> getRecordFn;
+  private MessageReceiver messageReceiver = null;
+  private final int minMessagesReceived;
+
+  public MockSessionService(
+      SerializableFunction<Integer, BytesXMLMessage> getRecordFn, int minMessagesReceived) {
+    this.getRecordFn = getRecordFn;
+    this.minMessagesReceived = minMessagesReceived;
+  }
+
+  @Override
+  public void close() {}
+
+  @Override
+  public boolean isClosed() {
+    return false;
+  }
+
+  @Override
+  public MessageReceiver createReceiver() {
+    if (messageReceiver == null) {
+      messageReceiver = new MockReceiver(getRecordFn, minMessagesReceived);
+    }
+    return messageReceiver;
+  }
+
+  @Override
+  public void connect() {}
+
+  public static class MockReceiver implements MessageReceiver, Serializable {
+    private final AtomicInteger counter = new AtomicInteger();
+    private final SerializableFunction<Integer, BytesXMLMessage> getRecordFn;
+    private final int minMessagesReceived;
+
+    public MockReceiver(
+        SerializableFunction<Integer, BytesXMLMessage> getRecordFn, int minMessagesReceived) {
+      this.getRecordFn = getRecordFn;
+      this.minMessagesReceived = minMessagesReceived;
+    }
+
+    @Override
+    public void start() {}
+
+    @Override
+    public boolean isClosed() {
+      return false;
+    }
+
+    @Override
+    public BytesXMLMessage receive() throws IOException {
+      return getRecordFn.apply(counter.getAndIncrement());
+    }
+
+    @Override
+    public boolean isEOF() {
+      return counter.get() >= minMessagesReceived;
+    }
+  }
+}

sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/MockSessionServiceFactory.java

@@ -0,0 +1,38 @@
+/*
+ * 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.solace;
+
+import org.apache.beam.sdk.io.solace.broker.SessionService;
+import org.apache.beam.sdk.io.solace.broker.SessionServiceFactory;
+
+public class MockSessionServiceFactory extends SessionServiceFactory {
+  SessionService sessionService;
+
+  public MockSessionServiceFactory(SessionService clientService) {
+    this.sessionService = clientService;
+  }
+
+  public static SessionServiceFactory getDefaultMock() {
+    return new MockSessionServiceFactory(new MockEmptySessionService());
+  }
+
+  @Override
+  public SessionService create() {
+    return sessionService;
+  }
+}