bzablocki commented on code in PR #31476: URL: https://github.com/apache/beam/pull/31476#discussion_r1638638982
########## sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/broker/SessionServiceFactory.java: ########## @@ -0,0 +1,52 @@ +/* + * 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.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; Review Comment: Yes, this has to be marked as @Nullable. The SessionServiceFactory will be initialized when the pipeline is constructed, as one of the arguments to the SolaceIO. But the `queue` can only be set later, in the expand method. Consider the scenario when a user creates a pipeline and specifies to read from a Topic - in this case we need to create a Queue (which is like a Subscription in PubSub) that will read from this Topic ``` Queue initializedQueue = queueFromOptions != null ? queueFromOptions : initializeQueueForTopic(jobName, sempClientFactory); ``` only then we can store the queue reference in the SessionServiceFactory ``` sessionServiceFactory.setQueue(initializedQueue); ``` I was trying to avoid it when designing it, but I couldn't find a better way around it. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
