-
Notifications
You must be signed in to change notification settings - Fork 4.2k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* wip solace connector * wip solace connector * some checker errors resolved * all checker errors resolved * improving unit tests * respond to pr commments * Documentation * Small refactor - move data classes out of the client * refactor * Add github action for integration test of Solace * testing github workflow * bump testcontainers to 1.19.7 - soalce semp was updated with an admin user access * Use FlowHandle to acknowledge messages to make SolaceCheckpointMark's fields serializable. * Handle StaleSessionException error * Add @internal annotation to mark the SolaceIO API beta and subject to change. * Improve documentation * Back to ack based on bytesxmlmessages. Deduplicate default to false. * update changes.md with Solace read connector * remove ack by id code * remove todo comment * Add licenses to package-info.java files * Restructure documentation * update aws test after upgrading testcontainers version. * Disable publishing docs until the first pass on the master branch * Remove files from this branch to split PR into smaller chunks * refactor tests for readability * revert upgrade of testcontainers - not needed in this PR chunk * revert upgrade of testcontainers - not needed in this PR chunk * spotless * remove IT tests from this pr * Tech Writer review * Add a field to Solace.Record mapped from BytesXMLMessage.getAttachmentByteBuffer() * Add and fix some documentation * Remove CheckpointMark's reference to the UnboundedSolaceReader - unnecessary. * Revert "Remove CheckpointMark's reference to the UnboundedSolaceReader - unnecessary." This reverts commit 2e1c10e. * Solace project init - github workflow file, gradle module * Splitting the #31476 - Leaving only PTransform AutoValue configurations * remove unnecessary dependencies * remove info from CHANGES.md * Add watermark-related code * Remove excessive @nullable annotations on Solace.Record class * Remove entry from CHANGES.md * Make Record builder package-private * Refactor SolaceIO - the constructor of Read takes a configuration builder as an argument * Change payload and attachments type to immutable ByteString * Downgrade Record builders access modifiers to package private * Add documentation * Add documentation to classes and methods in Solace.java * typo * Add SolaceCheckpointMark.java * Make SolaceCheckpointMark visible for testing * Remove SolaceRecordCoder and take advantage of @DefaultSchema
- Loading branch information
Showing
11 changed files
with
2,008 additions
and
40 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
415 changes: 375 additions & 40 deletions
415
sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/SolaceIO.java
Large diffs are not rendered by default.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
46 changes: 46 additions & 0 deletions
46
sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/MockEmptySessionService.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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); | ||
} | ||
} |
87 changes: 87 additions & 0 deletions
87
sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/MockSempClient.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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); | ||
} | ||
} |
38 changes: 38 additions & 0 deletions
38
sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/MockSempClientFactory.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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; | ||
} | ||
} |
88 changes: 88 additions & 0 deletions
88
sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/MockSessionService.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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; | ||
} | ||
} | ||
} |
38 changes: 38 additions & 0 deletions
38
...java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/MockSessionServiceFactory.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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; | ||
} | ||
} |
Oops, something went wrong.