Skip to content

Commit 2aeaaff

Browse files
KeranYangKeranYang
authored
chore: add building examples as part of CI (#184)
Signed-off-by: Keran Yang <[email protected]>
1 parent 57e4214 commit 2aeaaff

File tree

6 files changed

+119
-83
lines changed

6 files changed

+119
-83
lines changed

‎.github/workflows/run-tests.yaml

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@ jobs:
1717
distribution: 'temurin'
1818
- name: Build with Maven, run unit tests and the coverage check
1919
run: mvn clean install
20+
- name: Build Examples
21+
run: cd examples && mvn clean install
2022
- name: Upload coverage reports to Codecov
2123
uses: codecov/codecov-action@v4
2224
with:

‎examples/.editorconfig

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
root = true
2+
3+
[*]
4+
charset = utf-8
5+
end_of_line = lf
6+
insert_final_newline = true
7+
max_line_length = 100
8+
9+
[*.java]
10+
indent_size = 4
11+
indent_style = space
12+
tab_width = 4
13+
ij_continuation_indent_size = 8
14+
ij_java_binary_operation_sign_on_next_line = true
15+
ij_java_binary_operation_wrap = normal
16+
ij_java_call_parameters_new_line_after_left_paren = true
17+
ij_java_call_parameters_wrap = on_every_item
18+
ij_java_class_count_to_use_import_on_demand = 9999
19+
ij_java_doc_add_blank_line_after_param_comments = true
20+
ij_java_doc_add_blank_line_after_return = true
21+
ij_java_doc_align_exception_comments = false
22+
ij_java_doc_align_param_comments = false
23+
ij_java_doc_do_not_wrap_if_one_line = true
24+
ij_java_doc_enable_formatting = true
25+
ij_java_doc_indent_on_continuation = true
26+
ij_java_doc_keep_empty_lines = true
27+
ij_java_doc_preserve_line_breaks = false
28+
ij_java_layout_static_imports_separately = true
29+
ij_java_method_call_chain_wrap = on_every_item
30+
ij_java_method_parameters_new_line_after_left_paren = true
31+
ij_java_method_parameters_wrap = on_every_item
32+
ij_java_names_count_to_use_import_on_demand = 9999
33+
ij_java_spaces_around_equality_operators = true
34+
ij_java_variable_annotation_wrap = normal
35+
ij_java_wrap_first_method_in_call_chain = true

‎examples/src/test/java/io/numaproj/numaflow/examples/server/ServerTest.java

Lines changed: 70 additions & 68 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,6 @@
2929
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
3030
@Slf4j
3131
public class ServerTest {
32-
3332
@Test
3433
@Order(1)
3534
public void testMapServerInvocation() {
@@ -73,10 +72,8 @@ public void testFlatMapServerInvocation() {
7372
}
7473

7574
MapperTestKit.Client client = new MapperTestKit.Client();
76-
MapperTestKit.TestDatum datum = MapperTestKit.TestDatum
77-
.builder()
78-
.value("apple,banana,carrot".getBytes())
79-
.build();
75+
MapperTestKit.TestDatum datum =
76+
MapperTestKit.TestDatum.builder().value("apple,banana,carrot".getBytes()).build();
8077

8178
MessageList result = client.sendRequest(new String[]{}, datum);
8279

@@ -113,26 +110,27 @@ public void testReduceServerInvocation() {
113110
// create 10 datum with values 1 to 10
114111
List<Datum> datumList = new ArrayList<>();
115112
for (int i = 1; i <= 10; i++) {
116-
datumList.add(ReducerTestKit.TestDatum
117-
.builder()
118-
.value(Integer.toString(i).getBytes())
119-
.build());
113+
datumList.add(
114+
ReducerTestKit.TestDatum
115+
.builder()
116+
.value(Integer.toString(i).getBytes())
117+
.build());
120118
}
121119

122120
// create a client and send requests to the server
123121
ReducerTestKit.Client client = new ReducerTestKit.Client();
124122

125-
ReducerTestKit.TestReduceRequest testReduceRequest = ReducerTestKit.TestReduceRequest
126-
.builder()
127-
.datumList(datumList)
128-
.keys(new String[]{"test-key"})
129-
.startTime(Instant.ofEpochSecond(60000))
130-
.endTime(Instant.ofEpochSecond(60010))
131-
.build();
123+
ReducerTestKit.TestReduceRequest testReduceRequest =
124+
ReducerTestKit.TestReduceRequest.builder()
125+
.datumList(datumList)
126+
.keys(new String[]{"test-key"})
127+
.startTime(Instant.ofEpochSecond(60000))
128+
.endTime(Instant.ofEpochSecond(60010))
129+
.build();
132130

133131
try {
134-
io.numaproj.numaflow.reducer.MessageList messageList = client.sendReduceRequest(
135-
testReduceRequest);
132+
io.numaproj.numaflow.reducer.MessageList messageList =
133+
client.sendReduceRequest(testReduceRequest);
136134
// check if the response is correct
137135
if (messageList.getMessages().size() != 1) {
138136
Assertions.fail("Expected 1 message in the response");
@@ -169,12 +167,12 @@ public void testSinkServerInvocation() {
169167
// Create a test datum iterator with 10 messages
170168
SinkerTestKit.TestListIterator testListIterator = new SinkerTestKit.TestListIterator();
171169
for (int i = 0; i < datumCount; i++) {
172-
testListIterator.addDatum(SinkerTestKit.TestDatum
173-
.builder()
174-
.id("id-" + i)
175-
.value(("value-" + i).getBytes())
176-
.headers(Map.of("test-key", "test-value"))
177-
.build());
170+
testListIterator.addDatum(
171+
SinkerTestKit.TestDatum.builder()
172+
.id("id-" + i)
173+
.value(("value-" + i).getBytes())
174+
.headers(Map.of("test-key", "test-value"))
175+
.build());
178176
}
179177

180178
SinkerTestKit.Client client = new SinkerTestKit.Client();
@@ -199,47 +197,51 @@ public void testSinkServerInvocation() {
199197
// we can add the logic to verify if the messages were
200198
// successfully written to the sink(could be a file, database, etc.)
201199
}
202-
// FIXME: once tester kit changes are done for bidirectional streaming source
203-
// @Ignore
204-
// @Test
205-
// @Order(5)
206-
// public void testSourceServerInvocation() {
207-
// SimpleSource simpleSource = new SimpleSource();
208-
//
209-
// SourcerTestKit sourcerTestKit = new SourcerTestKit(simpleSource);
210-
// try {
211-
// sourcerTestKit.startServer();
212-
// } catch (Exception e) {
213-
// Assertions.fail("Failed to start server");
214-
// }
215-
//
216-
// // create a client to send requests to the server
217-
// SourcerTestKit.Client sourcerClient = new SourcerTestKit.Client();
218-
// // create a test observer to receive messages from the server
219-
// SourcerTestKit.TestListBasedObserver testObserver = new SourcerTestKit.TestListBasedObserver();
220-
// // create a read request with count 10 and timeout 1 second
221-
// SourcerTestKit.TestReadRequest testReadRequest = SourcerTestKit.TestReadRequest.builder()
222-
// .count(10).timeout(Duration.ofSeconds(1)).build();
223-
//
224-
// try {
225-
// sourcerClient.sendReadRequest(testReadRequest, testObserver);
226-
// Assertions.assertEquals(10, testObserver.getMessages().size());
227-
// } catch (Exception e) {
228-
// Assertions.fail("Failed to send request to server");
229-
// }
230-
//
231-
// try {
232-
// sourcerClient.close();
233-
// sourcerTestKit.stopServer();
234-
// } catch (InterruptedException e) {
235-
// Assertions.fail("Failed to stop server");
236-
// }
237-
// }
200+
201+
// FIXME: once tester kit changes are done for bidirectional streaming source
202+
// @Ignore
203+
// @Test
204+
// @Order(5)
205+
// public void testSourceServerInvocation() {
206+
// SimpleSource simpleSource = new SimpleSource();
207+
//
208+
// SourcerTestKit sourcerTestKit = new SourcerTestKit(simpleSource);
209+
// try {
210+
// sourcerTestKit.startServer();
211+
// } catch (Exception e) {
212+
// Assertions.fail("Failed to start server");
213+
// }
214+
//
215+
// // create a client to send requests to the server
216+
// SourcerTestKit.Client sourcerClient = new SourcerTestKit.Client();
217+
// // create a test observer to receive messages from the server
218+
// SourcerTestKit.TestListBasedObserver testObserver = new
219+
// SourcerTestKit.TestListBasedObserver();
220+
// // create a read request with count 10 and timeout 1 second
221+
// SourcerTestKit.TestReadRequest testReadRequest =
222+
// SourcerTestKit.TestReadRequest.builder()
223+
// .count(10).timeout(Duration.ofSeconds(1)).build();
224+
//
225+
// try {
226+
// sourcerClient.sendReadRequest(testReadRequest, testObserver);
227+
// Assertions.assertEquals(10, testObserver.getMessages().size());
228+
// } catch (Exception e) {
229+
// Assertions.fail("Failed to send request to server");
230+
// }
231+
//
232+
// try {
233+
// sourcerClient.close();
234+
// sourcerTestKit.stopServer();
235+
// } catch (InterruptedException e) {
236+
// Assertions.fail("Failed to stop server");
237+
// }
238+
// }
238239

239240
@Test
240241
@Order(6)
241242
public void testSourceTransformerServerInvocation() {
242-
SourceTransformerTestKit sourceTransformerTestKit = new SourceTransformerTestKit(new EventTimeFilterFunction());
243+
SourceTransformerTestKit sourceTransformerTestKit =
244+
new SourceTransformerTestKit(new EventTimeFilterFunction());
243245
try {
244246
sourceTransformerTestKit.startServer();
245247
} catch (Exception e) {
@@ -249,13 +251,13 @@ public void testSourceTransformerServerInvocation() {
249251
// Create a client which can send requests to the server
250252
SourceTransformerTestKit.Client client = new SourceTransformerTestKit.Client();
251253

252-
SourceTransformerTestKit.TestDatum datum = SourceTransformerTestKit.TestDatum.builder()
253-
.eventTime(Instant.ofEpochMilli(1640995200000L))
254-
.value("test".getBytes())
255-
.build();
256-
io.numaproj.numaflow.sourcetransformer.MessageList result = client.sendRequest(
257-
new String[]{},
258-
datum);
254+
SourceTransformerTestKit.TestDatum datum =
255+
SourceTransformerTestKit.TestDatum.builder()
256+
.eventTime(Instant.ofEpochMilli(1640995200000L))
257+
.value("test".getBytes())
258+
.build();
259+
io.numaproj.numaflow.sourcetransformer.MessageList result =
260+
client.sendRequest(new String[]{}, datum);
259261

260262
List<io.numaproj.numaflow.sourcetransformer.Message> messages = result.getMessages();
261263
Assertions.assertEquals(1, messages.size());

‎examples/src/test/java/io/numaproj/numaflow/examples/sink/simple/SimpleSinkTest.java

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -7,25 +7,23 @@
77
import org.junit.jupiter.api.Assertions;
88
import org.junit.jupiter.api.Test;
99

10-
1110
@Slf4j
1211
public class SimpleSinkTest {
1312

1413
@Test
1514
public void testSimpleSink() {
1615
int datumCount = 10;
1716
SimpleSink simpleSink = new SimpleSink();
18-
1917
// Create a test datum iterator with 10 messages
2018
SinkerTestKit.TestListIterator testListIterator = new SinkerTestKit.TestListIterator();
2119
for (int i = 0; i < datumCount; i++) {
22-
testListIterator.addDatum(SinkerTestKit.TestDatum
23-
.builder()
24-
.id("id-" + i)
25-
.value(("value-" + i).getBytes())
26-
.build());
20+
testListIterator.addDatum(
21+
SinkerTestKit.TestDatum
22+
.builder()
23+
.id("id-" + i)
24+
.value(("value-" + i).getBytes())
25+
.build());
2726
}
28-
2927
ResponseList responseList = simpleSink.processMessages(testListIterator);
3028
Assertions.assertEquals(datumCount, responseList.getResponses().size());
3129
for (Response response : responseList.getResponses()) {

‎src/main/java/io/numaproj/numaflow/sinker/Response.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -13,9 +13,9 @@
1313
@AllArgsConstructor(access = AccessLevel.PRIVATE)
1414
public class Response {
1515
private final String id;
16-
private final boolean success;
16+
private final Boolean success;
1717
private final String err;
18-
private final boolean fallback;
18+
private final Boolean fallback;
1919

2020
/**
2121
* Static method to create response for successful message processing.

‎src/main/java/io/numaproj/numaflow/sinker/Service.java

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -10,15 +10,14 @@
1010
import io.numaproj.numaflow.shared.ExceptionUtils;
1111
import io.numaproj.numaflow.sink.v1.SinkGrpc;
1212
import io.numaproj.numaflow.sink.v1.SinkOuterClass;
13-
import lombok.AllArgsConstructor;
14-
import lombok.extern.slf4j.Slf4j;
15-
1613
import java.time.Instant;
1714
import java.util.List;
1815
import java.util.concurrent.CompletableFuture;
1916
import java.util.concurrent.ExecutorService;
2017
import java.util.concurrent.Executors;
2118
import java.util.concurrent.TimeUnit;
19+
import lombok.AllArgsConstructor;
20+
import lombok.extern.slf4j.Slf4j;
2221

2322
@Slf4j
2423
@AllArgsConstructor
@@ -139,8 +138,8 @@ public void onCompleted() {
139138
}
140139

141140
private SinkOuterClass.SinkResponse.Result buildResult(Response response) {
142-
SinkOuterClass.Status status = response.isFallback() ? SinkOuterClass.Status.FALLBACK
143-
: response.isSuccess() ? SinkOuterClass.Status.SUCCESS : SinkOuterClass.Status.FAILURE;
141+
SinkOuterClass.Status status = response.getFallback() ? SinkOuterClass.Status.FALLBACK
142+
: response.getSuccess() ? SinkOuterClass.Status.SUCCESS : SinkOuterClass.Status.FAILURE;
144143
return SinkOuterClass.SinkResponse.Result.newBuilder()
145144
.setId(response.getId() == null ? "" : response.getId())
146145
.setErrMsg(response.getErr() == null ? "" : response.getErr())

0 commit comments

Comments
 (0)