fix issues

Author: liyubin117

flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/ElasticsearchRowDataLookupFunction.java

@@ -28,6 +28,7 @@
 import org.apache.flink.table.functions.FunctionContext;
 import org.apache.flink.table.functions.LookupFunction;
 import org.apache.flink.table.types.DataType;
+import org.apache.flink.util.FlinkRuntimeException;
 import org.apache.flink.util.Preconditions;
 
 import org.elasticsearch.action.search.SearchRequest;
@@ -49,7 +50,7 @@
 
 import static org.apache.flink.util.Preconditions.checkNotNull;
 
-/** A lookup function implementing {@link LookupTableSource} in elasticsearch connector. */
+/** A lookup function implementing {@link LookupTableSource} in Elasticsearch connector. */
 @Internal
 public class ElasticsearchRowDataLookupFunction<C extends AutoCloseable> extends LookupFunction {
 
@@ -155,14 +156,14 @@ public Collection<RowData> lookup(RowData keyRow) {
             } catch (IOException e) {
                 LOG.error(String.format("Elasticsearch search error, retry times = %d", retry), e);
                 if (retry >= maxRetryTimes) {
-                    throw new RuntimeException("Execution of Elasticsearch search failed.", e);
+                    throw new FlinkRuntimeException("Execution of Elasticsearch search failed.", e);
                 }
                 try {
                     Thread.sleep(1000L * retry);
                 } catch (InterruptedException e1) {
                     LOG.warn(
                             "Interrupted while waiting to retry failed elasticsearch search, aborting");
-                    throw new RuntimeException(e1);
+                    throw new FlinkRuntimeException(e1);
                 }
             }
         }

flink-connector-elasticsearch6/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch6DynamicSource.java

@@ -120,7 +120,7 @@ public DynamicTableSource copy() {
 
     @Override
     public String asSummaryString() {
-        return "Elasticsearch-6";
+        return "Elasticsearch6";
     }
 
     @Override

flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7DynamicSource.java

@@ -83,7 +83,7 @@ public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) {
                             config.getPathPrefix().orElse(null));
         }
 
-        Elasticsearch7ApiCallBridge elasticsearch6ApiCallBridge =
+        Elasticsearch7ApiCallBridge elasticsearch7ApiCallBridge =
                 new Elasticsearch7ApiCallBridge(config.getHosts(), restClientFactory);
 
         // Elasticsearch only support non-nested look up keys
@@ -104,7 +104,7 @@ public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) {
                         DataType.getFieldNames(physicalRowDataType).toArray(new String[0]),
                         DataType.getFieldDataTypes(physicalRowDataType).toArray(new DataType[0]),
                         keyNames,
-                        elasticsearch6ApiCallBridge);
+                        elasticsearch7ApiCallBridge);
         if (lookupCache != null) {
             return PartialCachingLookupProvider.of(lookupFunction, lookupCache);
         } else {
@@ -120,7 +120,7 @@ public DynamicTableSource copy() {
 
     @Override
     public String asSummaryString() {
-        return "Elasticsearch-6";
+        return "Elasticsearch7";
     }
 
     @Override