forked from apache/flink-connector-opensearch
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
21 changed files
with
551 additions
and
204 deletions.
There are no files selected for viewing
2 changes: 1 addition & 1 deletion
2
flink-connector-opensearch-base/archunit-violations/5c4a6228-f9cb-4828-9625-43c57d133967
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 |
|---|---|---|
| @@ -1,3 +1,3 @@ | ||
| Constructor <org.apache.flink.connector.opensearch.sink.BulkProcessorConfig.<init>(int, int, long, org.apache.flink.connector.opensearch.sink.FlushBackoffType, int, long)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (BulkProcessorConfig.java:44) | ||
| Constructor <org.apache.flink.connector.opensearch.table.OpensearchConfiguration.<init>(org.apache.flink.configuration.ReadableConfig)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (OpensearchConfiguration.java:61) | ||
| Method <org.apache.flink.connector.opensearch.table.IndexGeneratorFactory.createRuntimeIndexGenerator(java.lang.String, [Ljava.lang.String;, [Lorg.apache.flink.table.types.DataType;, org.apache.flink.connector.opensearch.table.IndexGeneratorFactory$IndexHelper, java.time.ZoneId)> has parameter of type <[Lorg.apache.flink.table.types.DataType;> in (IndexGeneratorFactory.java:0) | ||
| Method <org.apache.flink.connector.opensearch.table.IndexGeneratorFactory.createRuntimeIndexGenerator(java.lang.String, [Ljava.lang.String;, [Lorg.apache.flink.table.types.DataType;, org.apache.flink.connector.opensearch.table.IndexGeneratorFactory$IndexHelper, java.time.ZoneId)> has parameter of type <[Lorg.apache.flink.table.types.DataType;> in (IndexGeneratorFactory.java:0) |
File renamed without changes.
14 changes: 14 additions & 0 deletions
14
...-base/src/main/java/org/apache/flink/connector/opensearch/sink/DefaultFailureHandler.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,14 @@ | ||
| package org.apache.flink.connector.opensearch.sink; | ||
|
|
||
| import org.apache.flink.util.FlinkRuntimeException; | ||
|
|
||
| class DefaultFailureHandler implements FailureHandler { | ||
|
|
||
| @Override | ||
| public void onFailure(Throwable failure) { | ||
| if (failure instanceof FlinkRuntimeException) { | ||
| throw (FlinkRuntimeException) failure; | ||
| } | ||
| throw new FlinkRuntimeException(failure); | ||
| } | ||
| } |
File renamed without changes.
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
66 changes: 33 additions & 33 deletions
66
flink-connector-opensearch/archunit-violations/5c4a6228-f9cb-4828-9625-43c57d133967
Large diffs are not rendered by default.
Oops, something went wrong.
91 changes: 91 additions & 0 deletions
91
...rc/main/java/org/apache/flink/connector/opensearch/sink/DefaultBulkResponseInspector.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,91 @@ | ||
| /* | ||
| * 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.flink.connector.opensearch.sink; | ||
|
|
||
| import org.apache.flink.annotation.VisibleForTesting; | ||
| import org.apache.flink.util.FlinkRuntimeException; | ||
|
|
||
| import org.opensearch.action.DocWriteRequest; | ||
| import org.opensearch.action.bulk.BulkItemResponse; | ||
| import org.opensearch.action.bulk.BulkRequest; | ||
| import org.opensearch.action.bulk.BulkResponse; | ||
| import org.opensearch.rest.RestStatus; | ||
|
|
||
| import static org.apache.flink.util.ExceptionUtils.firstOrSuppressed; | ||
| import static org.apache.flink.util.Preconditions.checkNotNull; | ||
|
|
||
| /** | ||
| * A strict implementation that fails if either the whole bulk request failed or any of its actions. | ||
| */ | ||
| class DefaultBulkResponseInspector implements BulkResponseInspector { | ||
|
|
||
| @VisibleForTesting final FailureHandler failureHandler; | ||
|
|
||
| DefaultBulkResponseInspector() { | ||
| this(new DefaultFailureHandler()); | ||
| } | ||
|
|
||
| DefaultBulkResponseInspector(FailureHandler failureHandler) { | ||
| this.failureHandler = checkNotNull(failureHandler); | ||
| } | ||
|
|
||
| @Override | ||
| public void inspect(BulkRequest request, BulkResponse response) { | ||
| if (!response.hasFailures()) { | ||
| return; | ||
| } | ||
|
|
||
| Throwable chainedFailures = null; | ||
| for (int i = 0; i < response.getItems().length; i++) { | ||
| final BulkItemResponse itemResponse = response.getItems()[i]; | ||
| if (!itemResponse.isFailed()) { | ||
| continue; | ||
| } | ||
| final Throwable failure = itemResponse.getFailure().getCause(); | ||
| if (failure == null) { | ||
| continue; | ||
| } | ||
| final RestStatus restStatus = itemResponse.getFailure().getStatus(); | ||
| final DocWriteRequest<?> actionRequest = request.requests().get(i); | ||
|
|
||
| chainedFailures = | ||
| firstOrSuppressed( | ||
| wrapException(restStatus, failure, actionRequest), chainedFailures); | ||
| } | ||
| if (chainedFailures == null) { | ||
| return; | ||
| } | ||
| failureHandler.onFailure(chainedFailures); | ||
| } | ||
|
|
||
| private static Throwable wrapException( | ||
| RestStatus restStatus, Throwable rootFailure, DocWriteRequest<?> actionRequest) { | ||
| if (restStatus == null) { | ||
| return new FlinkRuntimeException( | ||
| String.format("Single action %s of bulk request failed.", actionRequest), | ||
| rootFailure); | ||
| } else { | ||
| return new FlinkRuntimeException( | ||
| String.format( | ||
| "Single action %s of bulk request failed with status %s.", | ||
| actionRequest, restStatus.getStatus()), | ||
| rootFailure); | ||
| } | ||
| } | ||
| } |
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
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
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
8 changes: 4 additions & 4 deletions
8
flink-connector-opensearch2/archunit-violations/4382f1f0-807a-45ff-97d8-42f72b6e9484
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 |
|---|---|---|
| @@ -1,18 +1,18 @@ | ||
| org.apache.flink.connector.opensearch.sink.Opensearch2SinkITCase does not satisfy: only one of the following predicates match:\ | ||
| * reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\ | ||
| * reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension\ | ||
| * reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension or are , and of type MiniClusterTestEnvironment and annotated with @TestEnv\ | ||
| * reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\ | ||
| * reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\ | ||
| or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule | ||
| org.apache.flink.connector.opensearch.sink.Opensearch2WriterITCase does not satisfy: only one of the following predicates match:\ | ||
| * reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\ | ||
| * reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension\ | ||
| * reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension or are , and of type MiniClusterTestEnvironment and annotated with @TestEnv\ | ||
| * reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\ | ||
| * reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\ | ||
| or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule | ||
| org.apache.flink.connector.opensearch.table.Opensearch2DynamicSinkITCase does not satisfy: only one of the following predicates match:\ | ||
| * reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\ | ||
| * reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension\ | ||
| * reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension or are , and of type MiniClusterTestEnvironment and annotated with @TestEnv\ | ||
| * reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\ | ||
| * reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\ | ||
| or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule | ||
| or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule |
Oops, something went wrong.