This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch v1.2
in repository https://gitbox.apache.org/repos/asf/flink-connector-opensearch.git
The following commit(s) were added to refs/heads/v1.2 by this push:
new 5901d58 [hotfix] [opensearch] Make branch compatible with Flink
1.19/1.20
5901d58 is described below
commit 5901d58099157df9cb7d29bf33c27d98b9efd7e1
Author: Andriy Redko <[email protected]>
AuthorDate: Fri Sep 18 12:58:48 2026 -0400
[hotfix] [opensearch] Make branch compatible with Flink 1.19/1.20
Signed-off-by: Andriy Redko <[email protected]>
---
flink-connector-opensearch-e2e-tests/pom.xml | 1 +
.../streaming/tests/OpensearchSinkE2ECase.java | 9 ++++++-
.../opensearch/sink/OpensearchSinkBuilderTest.java | 21 ++-------------
.../opensearch/sink/OpensearchWriterITCase.java | 31 ++--------------------
.../streaming/tests/OpensearchSinkE2ECase.java | 9 ++++++-
.../opensearch/sink/Opensearch2WriterITCase.java | 31 ++--------------------
pom.xml | 7 -----
7 files changed, 23 insertions(+), 86 deletions(-)
diff --git a/flink-connector-opensearch-e2e-tests/pom.xml
b/flink-connector-opensearch-e2e-tests/pom.xml
index b0caa2e..2ad2eb2 100644
--- a/flink-connector-opensearch-e2e-tests/pom.xml
+++ b/flink-connector-opensearch-e2e-tests/pom.xml
@@ -64,6 +64,7 @@ under the License.
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-test-utils</artifactId>
+ <version>${flink.version}</version>
<scope>test</scope>
</dependency>
<dependency>
diff --git
a/flink-connector-opensearch-e2e-tests/src/test/java/org/apache/flink/streaming/tests/OpensearchSinkE2ECase.java
b/flink-connector-opensearch-e2e-tests/src/test/java/org/apache/flink/streaming/tests/OpensearchSinkE2ECase.java
index fe7217a..4e13fc9 100644
---
a/flink-connector-opensearch-e2e-tests/src/test/java/org/apache/flink/streaming/tests/OpensearchSinkE2ECase.java
+++
b/flink-connector-opensearch-e2e-tests/src/test/java/org/apache/flink/streaming/tests/OpensearchSinkE2ECase.java
@@ -19,6 +19,7 @@ package org.apache.flink.streaming.tests;
import org.apache.flink.connector.opensearch.test.DockerImageVersions;
import
org.apache.flink.connector.testframe.container.FlinkContainerTestEnvironment;
+import org.apache.flink.connector.testframe.container.FlinkContainersSettings;
import
org.apache.flink.connector.testframe.external.DefaultContainerizedExternalSystem;
import org.apache.flink.connector.testframe.external.ExternalSystemDataReader;
import org.apache.flink.connector.testframe.junit.annotations.TestContext;
@@ -52,7 +53,13 @@ public class OpensearchSinkE2ECase extends
SinkTestSuiteBase<ComparableTuple2<In
@TestSemantics
CheckpointingMode[] semantics = new CheckpointingMode[]
{CheckpointingMode.EXACTLY_ONCE};
- @TestEnv FlinkContainerTestEnvironment flink = new
FlinkContainerTestEnvironment(1, 6);
+ @TestEnv
+ FlinkContainerTestEnvironment flink =
+ FlinkContainerTestEnvironment.fromSettings(
+ FlinkContainersSettings.builder()
+ .numTaskManagers(1)
+ .numSlotsPerTaskManager(6)
+ .build());
public OpensearchSinkE2ECase() throws Exception {}
diff --git
a/flink-connector-opensearch/src/test/java/org/apache/flink/connector/opensearch/sink/OpensearchSinkBuilderTest.java
b/flink-connector-opensearch/src/test/java/org/apache/flink/connector/opensearch/sink/OpensearchSinkBuilderTest.java
index cf35411..fa2f14b 100644
---
a/flink-connector-opensearch/src/test/java/org/apache/flink/connector/opensearch/sink/OpensearchSinkBuilderTest.java
+++
b/flink-connector-opensearch/src/test/java/org/apache/flink/connector/opensearch/sink/OpensearchSinkBuilderTest.java
@@ -24,10 +24,8 @@ import
org.apache.flink.connector.opensearch.sink.BulkResponseInspector.BulkResp
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;
-import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.SimpleUserCodeClassLoader;
import org.apache.flink.util.TestLoggerExtension;
-import org.apache.flink.util.function.ThrowingRunnable;
import org.apache.http.HttpHost;
import org.junit.jupiter.api.DynamicTest;
@@ -163,8 +161,8 @@ class OpensearchSinkBuilderTest {
TestingSinkWriterMetricGroup.getSinkWriterMetricGroup(
new UnregisteredMetricsGroup()));
- Mockito.when(sinkInitContext.getMailboxExecutor())
- .thenReturn(new
OpensearchSinkBuilderTest.DummyMailboxExecutor());
+ MailboxExecutor executor = Mockito.mock(MailboxExecutor.class);
+
Mockito.when(sinkInitContext.getMailboxExecutor()).thenReturn(executor);
Mockito.when(sinkInitContext.getProcessingTimeService())
.thenReturn(new TestProcessingTimeService());
Mockito.when(sinkInitContext.getUserCodeClassLoader())
@@ -176,21 +174,6 @@ class OpensearchSinkBuilderTest {
assertThat(called).isTrue();
}
- private static class DummyMailboxExecutor implements MailboxExecutor {
- private DummyMailboxExecutor() {}
-
- public void execute(
- ThrowingRunnable<? extends Exception> command,
- String descriptionFormat,
- Object... descriptionArgs) {}
-
- public void yield() throws InterruptedException, FlinkRuntimeException
{}
-
- public boolean tryYield() throws FlinkRuntimeException {
- return false;
- }
- }
-
private OpensearchSinkBuilder<Object> createEmptyBuilder() {
return new OpensearchSinkBuilder<>();
}
diff --git
a/flink-connector-opensearch/src/test/java/org/apache/flink/connector/opensearch/sink/OpensearchWriterITCase.java
b/flink-connector-opensearch/src/test/java/org/apache/flink/connector/opensearch/sink/OpensearchWriterITCase.java
index fa7fc16..dd7c545 100644
---
a/flink-connector-opensearch/src/test/java/org/apache/flink/connector/opensearch/sink/OpensearchWriterITCase.java
+++
b/flink-connector-opensearch/src/test/java/org/apache/flink/connector/opensearch/sink/OpensearchWriterITCase.java
@@ -17,7 +17,6 @@
package org.apache.flink.connector.opensearch.sink;
-import org.apache.flink.api.common.operators.MailboxExecutor;
import org.apache.flink.api.connector.sink2.SinkWriter;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.opensearch.OpensearchUtil;
@@ -27,11 +26,10 @@ import org.apache.flink.metrics.Gauge;
import org.apache.flink.metrics.groups.OperatorIOMetricGroup;
import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
import org.apache.flink.metrics.testutils.MetricListener;
+import org.apache.flink.runtime.mailbox.SyncMailboxExecutor;
import org.apache.flink.runtime.metrics.MetricNames;
import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
-import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.TestLoggerExtension;
-import org.apache.flink.util.function.ThrowingRunnable;
import org.apache.http.HttpHost;
import org.junit.jupiter.api.AfterEach;
@@ -331,7 +329,7 @@ class OpensearchWriterITCase {
null,
true),
metricGroup,
- new TestMailbox(),
+ new SyncMailboxExecutor(),
new DefaultRestClientFactory(),
new DefaultBulkResponseInspector(failureHandler));
}
@@ -376,29 +374,4 @@ class OpensearchWriterITCase {
}
}
}
-
- private static class TestMailbox implements MailboxExecutor {
-
- @Override
- public void execute(
- ThrowingRunnable<? extends Exception> command,
- String descriptionFormat,
- Object... descriptionArgs) {
- try {
- command.run();
- } catch (Exception e) {
- throw new RuntimeException("Unexpected error", e);
- }
- }
-
- @Override
- public void yield() throws InterruptedException, FlinkRuntimeException
{
- Thread.sleep(100);
- }
-
- @Override
- public boolean tryYield() throws FlinkRuntimeException {
- return false;
- }
- }
}
diff --git
a/flink-connector-opensearch2-e2e-tests/src/test/java/org/apache/flink/streaming/tests/OpensearchSinkE2ECase.java
b/flink-connector-opensearch2-e2e-tests/src/test/java/org/apache/flink/streaming/tests/OpensearchSinkE2ECase.java
index eb14b0f..8b26f8b 100644
---
a/flink-connector-opensearch2-e2e-tests/src/test/java/org/apache/flink/streaming/tests/OpensearchSinkE2ECase.java
+++
b/flink-connector-opensearch2-e2e-tests/src/test/java/org/apache/flink/streaming/tests/OpensearchSinkE2ECase.java
@@ -19,6 +19,7 @@ package org.apache.flink.streaming.tests;
import org.apache.flink.connector.opensearch.test.DockerImageVersions;
import
org.apache.flink.connector.testframe.container.FlinkContainerTestEnvironment;
+import org.apache.flink.connector.testframe.container.FlinkContainersSettings;
import
org.apache.flink.connector.testframe.external.DefaultContainerizedExternalSystem;
import org.apache.flink.connector.testframe.external.ExternalSystemDataReader;
import org.apache.flink.connector.testframe.junit.annotations.TestContext;
@@ -52,7 +53,13 @@ public class OpensearchSinkE2ECase extends
SinkTestSuiteBase<ComparableTuple2<In
@TestSemantics
CheckpointingMode[] semantics = new CheckpointingMode[]
{CheckpointingMode.EXACTLY_ONCE};
- @TestEnv FlinkContainerTestEnvironment flink = new
FlinkContainerTestEnvironment(1, 6);
+ @TestEnv
+ FlinkContainerTestEnvironment flink =
+ FlinkContainerTestEnvironment.fromSettings(
+ FlinkContainersSettings.builder()
+ .numTaskManagers(1)
+ .numSlotsPerTaskManager(6)
+ .build());
public OpensearchSinkE2ECase() throws Exception {}
diff --git
a/flink-connector-opensearch2/src/test/java/org/apache/flink/connector/opensearch/sink/Opensearch2WriterITCase.java
b/flink-connector-opensearch2/src/test/java/org/apache/flink/connector/opensearch/sink/Opensearch2WriterITCase.java
index 92d0c09..b092d1a 100644
---
a/flink-connector-opensearch2/src/test/java/org/apache/flink/connector/opensearch/sink/Opensearch2WriterITCase.java
+++
b/flink-connector-opensearch2/src/test/java/org/apache/flink/connector/opensearch/sink/Opensearch2WriterITCase.java
@@ -17,7 +17,6 @@
package org.apache.flink.connector.opensearch.sink;
-import org.apache.flink.api.common.operators.MailboxExecutor;
import org.apache.flink.api.connector.sink2.SinkWriter;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.opensearch.OpensearchUtil;
@@ -27,11 +26,10 @@ import org.apache.flink.metrics.Gauge;
import org.apache.flink.metrics.groups.OperatorIOMetricGroup;
import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
import org.apache.flink.metrics.testutils.MetricListener;
+import org.apache.flink.runtime.mailbox.SyncMailboxExecutor;
import org.apache.flink.runtime.metrics.MetricNames;
import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
-import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.TestLoggerExtension;
-import org.apache.flink.util.function.ThrowingRunnable;
import org.apache.http.HttpHost;
import org.junit.jupiter.api.AfterEach;
@@ -331,7 +329,7 @@ class Opensearch2WriterITCase {
null,
true),
metricGroup,
- new TestMailbox(),
+ new SyncMailboxExecutor(),
new DefaultRestClientFactory(),
new DefaultBulkResponseInspector(failureHandler));
}
@@ -376,29 +374,4 @@ class Opensearch2WriterITCase {
}
}
}
-
- private static class TestMailbox implements MailboxExecutor {
-
- @Override
- public void execute(
- ThrowingRunnable<? extends Exception> command,
- String descriptionFormat,
- Object... descriptionArgs) {
- try {
- command.run();
- } catch (Exception e) {
- throw new RuntimeException("Unexpected error", e);
- }
- }
-
- @Override
- public void yield() throws InterruptedException, FlinkRuntimeException
{
- Thread.sleep(100);
- }
-
- @Override
- public boolean tryYield() throws FlinkRuntimeException {
- return false;
- }
- }
}
diff --git a/pom.xml b/pom.xml
index 6506e7d..bc9d195 100644
--- a/pom.xml
+++ b/pom.xml
@@ -296,13 +296,6 @@ under the License.
<version>3.14.0</version>
</dependency>
- <!-- For dependency convergence -->
- <dependency>
- <groupId>net.bytebuddy</groupId>
- <artifactId>byte-buddy</artifactId>
- <version>1.14.13</version>
- </dependency>
-
<!-- For dependency convergence -->
<dependency>
<groupId>com.fasterxml.jackson</groupId>
