Skip to content

Commit 925be57

Browse files
snuyanzinsnuyanzin
committed
[FLINK-34942] Add support for Flink 1.19, 1.20-SNAPSHOT
1 parent 9e161cc commit 925be57

File tree

10 files changed

+360
-93
lines changed

10 files changed

+360
-93
lines changed

‎.github/workflows/push_pr.yml

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,9 @@ jobs:
2525
compile_and_test:
2626
strategy:
2727
matrix:
28-
flink: [1.17.1, 1.18-SNAPSHOT]
28+
flink: [ 1.17-SNAPSHOT, 1.18-SNAPSHOT, 1.19-SNAPSHOT, 1.20-SNAPSHOT ]
29+
jdk: [ '8, 11' ]
2930
uses: apache/flink-connector-shared-utils/.github/workflows/ci.yml@ci_utils
3031
with:
3132
flink_version: ${{ matrix.flink }}
33+
jdk_version: ${{ matrix.jdk }}

‎.github/workflows/weekly.yml

Lines changed: 13 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -26,21 +26,31 @@ jobs:
2626
if: github.repository_owner == 'apache'
2727
strategy:
2828
matrix:
29-
flink_branches: [{
29+
flink_branches: [ {
3030
flink: 1.17-SNAPSHOT,
3131
branch: main
3232
}, {
3333
flink: 1.18-SNAPSHOT,
34+
jdk: '8, 11',
3435
branch: main
3536
}, {
36-
flink: 1.16.2,
37+
flink: 1.19-SNAPSHOT,
38+
jdk: '8, 11',
39+
branch: main
40+
}, {
41+
flink: 1.20-SNAPSHOT,
42+
jdk: '8, 11',
43+
branch: main
44+
}, {
45+
flink: 1.17.2,
3746
branch: v1.0
3847
}, {
39-
flink: 1.17.1,
48+
flink: 1.18.1,
4049
branch: v1.0
4150
}]
4251
uses: apache/flink-connector-shared-utils/.github/workflows/ci.yml@ci_utils
4352
with:
4453
flink_version: ${{ matrix.flink_branches.flink }}
54+
jdk_version: ${{ matrix.flink_branches.jdk || '8, 11' }}
4555
connector_branch: ${{ matrix.flink_branches.branch }}
4656
run_dependency_convergence: false

‎flink-connector-opensearch-e2e-tests/src/test/java/org/apache/flink/core/execution/CheckpointingMode.java

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,44 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.flink.core.execution;
19+
20+
import org.apache.flink.streaming.tests.OpensearchSinkE2ECase;
21+
22+
/**
23+
* This is a copy of {@link CheckpointingMode} from flink-core module introduced in Flink 1.20. We
24+
* need it here to make {@link OpensearchSinkE2ECase} compatible with earlier releases. Could be
25+
* removed together with dropping support of Flink 1.19.
26+
*/
27+
public enum CheckpointingMode {
28+
EXACTLY_ONCE,
29+
AT_LEAST_ONCE;
30+
31+
private CheckpointingMode() {}
32+
33+
public static org.apache.flink.streaming.api.CheckpointingMode convertFromCheckpointingMode(
34+
org.apache.flink.core.execution.CheckpointingMode semantic) {
35+
switch (semantic) {
36+
case EXACTLY_ONCE:
37+
return org.apache.flink.streaming.api.CheckpointingMode.EXACTLY_ONCE;
38+
case AT_LEAST_ONCE:
39+
return org.apache.flink.streaming.api.CheckpointingMode.AT_LEAST_ONCE;
40+
default:
41+
throw new IllegalArgumentException("Unsupported semantic: " + semantic);
42+
}
43+
}
44+
}

‎flink-connector-opensearch-e2e-tests/src/test/java/org/apache/flink/streaming/tests/OpensearchSinkE2ECase.java

Lines changed: 27 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@
4040
import java.util.List;
4141

4242
import static org.apache.flink.connector.testframe.utils.CollectIteratorAssertions.assertThat;
43+
import static org.apache.flink.core.execution.CheckpointingMode.convertFromCheckpointingMode;
4344
import static org.apache.flink.runtime.testutils.CommonTestUtils.waitUntilCondition;
4445

4546
/** End to end test for OpensearchSink based on connector testing framework. */
@@ -85,7 +86,8 @@ public OpensearchSinkE2ECase() throws Exception {}
8586
.toUri()
8687
.toURL()));
8788

88-
@Override
89+
/** Could be removed together with dropping support of Flink 1.19. */
90+
@Deprecated
8991
protected void checkResultWithSemantic(
9092
ExternalSystemDataReader<ComparableTuple2<Integer, String>> reader,
9193
List<ComparableTuple2<Integer, String>> testData,
@@ -109,6 +111,30 @@ protected void checkResultWithSemantic(
109111
READER_RETRY_ATTEMPTS);
110112
}
111113

114+
protected void checkResultWithSemantic(
115+
ExternalSystemDataReader<ComparableTuple2<Integer, String>> reader,
116+
List<ComparableTuple2<Integer, String>> testData,
117+
org.apache.flink.core.execution.CheckpointingMode semantic)
118+
throws Exception {
119+
waitUntilCondition(
120+
() -> {
121+
try {
122+
List<ComparableTuple2<Integer, String>> result =
123+
reader.poll(Duration.ofMillis(READER_TIMEOUT));
124+
assertThat(sort(result).iterator())
125+
.matchesRecordsFromSource(
126+
Collections.singletonList(sort(testData)),
127+
convertFromCheckpointingMode(semantic));
128+
return true;
129+
} catch (Throwable t) {
130+
LOG.warn("Polled results not as expected", t);
131+
return false;
132+
}
133+
},
134+
5000,
135+
READER_RETRY_ATTEMPTS);
136+
}
137+
112138
private static <T extends Comparable<T>> List<T> sort(List<T> list) {
113139
Collections.sort(list);
114140
return list;

‎flink-connector-opensearch/src/test/java/org/apache/flink/connector/opensearch/sink/OpensearchSinkBuilderTest.java

Lines changed: 20 additions & 70 deletions
Original file line numberDiff line numberDiff line change
@@ -17,25 +17,17 @@
1717

1818
package org.apache.flink.connector.opensearch.sink;
1919

20-
import org.apache.flink.api.common.JobID;
2120
import org.apache.flink.api.common.operators.MailboxExecutor;
22-
import org.apache.flink.api.common.operators.ProcessingTimeService;
23-
import org.apache.flink.api.common.serialization.SerializationSchema;
24-
import org.apache.flink.api.common.typeutils.TypeSerializer;
25-
import org.apache.flink.api.connector.sink2.Sink;
2621
import org.apache.flink.api.connector.sink2.Sink.InitContext;
2722
import org.apache.flink.connector.base.DeliveryGuarantee;
2823
import org.apache.flink.connector.opensearch.sink.BulkResponseInspector.BulkResponseInspectorFactory;
2924
import org.apache.flink.connector.opensearch.sink.OpensearchWriter.DefaultBulkResponseInspector;
3025
import org.apache.flink.metrics.MetricGroup;
31-
import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
3226
import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
33-
import org.apache.flink.runtime.metrics.groups.InternalSinkWriterMetricGroup;
3427
import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;
3528
import org.apache.flink.util.FlinkRuntimeException;
3629
import org.apache.flink.util.SimpleUserCodeClassLoader;
3730
import org.apache.flink.util.TestLoggerExtension;
38-
import org.apache.flink.util.UserCodeClassLoader;
3931
import org.apache.flink.util.function.ThrowingRunnable;
4032

4133
import org.apache.http.HttpHost;
@@ -44,8 +36,8 @@
4436
import org.junit.jupiter.api.TestFactory;
4537
import org.junit.jupiter.api.TestInstance;
4638
import org.junit.jupiter.api.extension.ExtendWith;
39+
import org.mockito.Mockito;
4740

48-
import java.util.OptionalLong;
4941
import java.util.concurrent.atomic.AtomicBoolean;
5042
import java.util.stream.Stream;
5143

@@ -138,9 +130,12 @@ void testOverrideFailureHandler() {
138130
final OpensearchSink<Object> sink =
139131
createMinimalBuilder().setFailureHandler(failureHandler).build();
140132

141-
final InitContext sinkInitContext = new MockInitContext();
142133
final BulkResponseInspector bulkResponseInspector =
143-
sink.getBulkResponseInspectorFactory().apply(sinkInitContext::metricGroup);
134+
sink.getBulkResponseInspectorFactory()
135+
.apply(
136+
() ->
137+
TestingSinkWriterMetricGroup.getSinkWriterMetricGroup(
138+
new UnregisteredMetricsGroup()));
144139
assertThat(bulkResponseInspector)
145140
.isInstanceOf(DefaultBulkResponseInspector.class)
146141
.extracting(
@@ -163,7 +158,20 @@ void testOverrideBulkResponseInspectorFactory() {
163158
.setBulkResponseInspectorFactory(bulkResponseInspectorFactory)
164159
.build();
165160

166-
final InitContext sinkInitContext = new MockInitContext();
161+
final InitContext sinkInitContext = Mockito.mock(InitContext.class);
162+
Mockito.when(sinkInitContext.metricGroup())
163+
.thenReturn(
164+
TestingSinkWriterMetricGroup.getSinkWriterMetricGroup(
165+
new UnregisteredMetricsGroup()));
166+
167+
Mockito.when(sinkInitContext.getMailboxExecutor())
168+
.thenReturn(new OpensearchSinkBuilderTest.DummyMailboxExecutor());
169+
Mockito.when(sinkInitContext.getProcessingTimeService())
170+
.thenReturn(new TestProcessingTimeService());
171+
Mockito.when(sinkInitContext.getUserCodeClassLoader())
172+
.thenReturn(
173+
SimpleUserCodeClassLoader.create(
174+
OpensearchSinkBuilderTest.class.getClassLoader()));
167175

168176
assertThatCode(() -> sink.createWriter(sinkInitContext)).doesNotThrowAnyException();
169177
assertThat(called).isTrue();
@@ -184,64 +192,6 @@ public boolean tryYield() throws FlinkRuntimeException {
184192
}
185193
}
186194

187-
private static class MockInitContext
188-
implements Sink.InitContext, SerializationSchema.InitializationContext {
189-
190-
public UserCodeClassLoader getUserCodeClassLoader() {
191-
return SimpleUserCodeClassLoader.create(
192-
OpensearchSinkBuilderTest.class.getClassLoader());
193-
}
194-
195-
public MailboxExecutor getMailboxExecutor() {
196-
return new OpensearchSinkBuilderTest.DummyMailboxExecutor();
197-
}
198-
199-
public ProcessingTimeService getProcessingTimeService() {
200-
return new TestProcessingTimeService();
201-
}
202-
203-
public int getSubtaskId() {
204-
return 0;
205-
}
206-
207-
public int getNumberOfParallelSubtasks() {
208-
return 0;
209-
}
210-
211-
public int getAttemptNumber() {
212-
return 0;
213-
}
214-
215-
public SinkWriterMetricGroup metricGroup() {
216-
return InternalSinkWriterMetricGroup.mock(new UnregisteredMetricsGroup());
217-
}
218-
219-
public MetricGroup getMetricGroup() {
220-
return this.metricGroup();
221-
}
222-
223-
public OptionalLong getRestoredCheckpointId() {
224-
return OptionalLong.empty();
225-
}
226-
227-
public SerializationSchema.InitializationContext
228-
asSerializationSchemaInitializationContext() {
229-
return this;
230-
}
231-
232-
public boolean isObjectReuseEnabled() {
233-
return false;
234-
}
235-
236-
public <IN> TypeSerializer<IN> createInputSerializer() {
237-
throw new UnsupportedOperationException();
238-
}
239-
240-
public JobID getJobId() {
241-
throw new UnsupportedOperationException();
242-
}
243-
}
244-
245195
private OpensearchSinkBuilder<Object> createEmptyBuilder() {
246196
return new OpensearchSinkBuilder<>();
247197
}

‎flink-connector-opensearch/src/test/java/org/apache/flink/connector/opensearch/sink/OpensearchWriterITCase.java

Lines changed: 11 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,6 @@
3030
import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
3131
import org.apache.flink.metrics.testutils.MetricListener;
3232
import org.apache.flink.runtime.metrics.MetricNames;
33-
import org.apache.flink.runtime.metrics.groups.InternalSinkWriterMetricGroup;
3433
import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
3534
import org.apache.flink.util.FlinkRuntimeException;
3635
import org.apache.flink.util.TestLoggerExtension;
@@ -168,17 +167,19 @@ void testIncrementByteOutMetric() throws Exception {
168167
final String index = "test-inc-byte-out";
169168
final OperatorIOMetricGroup operatorIOMetricGroup =
170169
UnregisteredMetricGroups.createUnregisteredOperatorMetricGroup().getIOMetricGroup();
171-
final InternalSinkWriterMetricGroup metricGroup =
172-
InternalSinkWriterMetricGroup.mock(
173-
metricListener.getMetricGroup(), operatorIOMetricGroup);
174170
final int flushAfterNActions = 2;
175171
final BulkProcessorConfig bulkProcessorConfig =
176172
new BulkProcessorConfig(flushAfterNActions, -1, -1, FlushBackoffType.NONE, 0, 0);
177173

178174
try (final OpensearchWriter<Tuple2<Integer, String>> writer =
179-
createWriter(index, false, bulkProcessorConfig, metricGroup)) {
175+
createWriter(
176+
index,
177+
false,
178+
bulkProcessorConfig,
179+
TestingSinkWriterMetricGroup.getSinkWriterMetricGroup(
180+
operatorIOMetricGroup, metricListener.getMetricGroup()))) {
180181
final Counter numBytesOut = operatorIOMetricGroup.getNumBytesOutCounter();
181-
assertThat(numBytesOut.getCount()).isEqualTo(0);
182+
assertThat(numBytesOut.getCount()).isZero();
182183
writer.write(Tuple2.of(1, buildMessage(1)), null);
183184
writer.write(Tuple2.of(2, buildMessage(2)), null);
184185

@@ -280,7 +281,8 @@ private OpensearchWriter<Tuple2<Integer, String>> createWriter(
280281
index,
281282
flushOnCheckpoint,
282283
bulkProcessorConfig,
283-
InternalSinkWriterMetricGroup.mock(metricListener.getMetricGroup()),
284+
TestingSinkWriterMetricGroup.getSinkWriterMetricGroup(
285+
metricListener.getMetricGroup()),
284286
new DefaultFailureHandler());
285287
}
286288

@@ -293,7 +295,8 @@ private OpensearchWriter<Tuple2<Integer, String>> createWriter(
293295
index,
294296
flushOnCheckpoint,
295297
bulkProcessorConfig,
296-
InternalSinkWriterMetricGroup.mock(metricListener.getMetricGroup()),
298+
TestingSinkWriterMetricGroup.getSinkWriterMetricGroup(
299+
metricListener.getMetricGroup()),
297300
failureHandler);
298301
}
299302

0 commit comments

Comments
 (0)