[hotfix] Fix some typos for connector-base, protobuf, runtime, table-planner and tests modules. (apache#28328)

Author: RocMarshal

flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSourceReader.java

@@ -223,7 +223,7 @@ private void setCurrentReader(int index) {
         try {
             reader = source.createReader(readerContext);
         } catch (Exception e) {
-            throw new RuntimeException("Failed tp create reader", e);
+            throw new RuntimeException("Failed to create reader", e);
         }
         // currentReader must be switched before `addSplits` is called.
         currentSourceIndex = index;

flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/util/PbSchemaValidationUtils.java

@@ -83,7 +83,7 @@ private static void validateTypeMatch(Descriptors.Descriptor descriptor, RowType
                                 throw new ValidationException(
                                         "Column "
                                                 + rowField.getName()
-                                                + " does not exists in definition of proto class.");
+                                                + " does not exist in definition of proto class.");
                             }
                         });
     }
@@ -160,7 +160,7 @@ private static void validateSimpleType(FieldDescriptor fd, LogicalTypeRoot logic
             throw new ValidationException(
                     "Protobuf field type does not match column type, "
                             + fd.getJavaType()
-                            + "(protobuf) is not compatible of "
+                            + "(protobuf) is not compatible with "
                             + logicalTypeRoot);
         }
     }

flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/DefaultApplicationStore.java

@@ -210,7 +210,7 @@ public void putApplication(ApplicationStoreEntry application) throws Exception {
                         success = true;
                     } catch (StateHandleStore.NotExistException ignored) {
                         LOG.warn(
-                                "{} does not exists in {}.",
+                                "{} does not exist in {}.",
                                 application,
                                 applicationStateHandleStore);
                     }

flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/DefaultExecutionPlanStore.java

@@ -235,7 +235,7 @@ public void putExecutionPlan(ExecutionPlan executionPlan) throws Exception {
                         success = true;
                     } catch (StateHandleStore.NotExistException ignored) {
                         LOG.warn(
-                                "{} does not exists in {}.",
+                                "{} does not exist in {}.",
                                 executionPlan,
                                 executionPlanStateHandleStore);
                     }

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/connectors/DynamicSinkUtils.java

@@ -515,7 +515,7 @@ private static RelNode convertUpdate(
                 && updateInfo.getRowLevelUpdateMode()
                         != SupportsRowLevelUpdate.RowLevelUpdateMode.ALL_ROWS) {
             throw new IllegalArgumentException(
-                    "Unknown update mode:" + updateInfo.getRowLevelUpdateMode());
+                    "Unknown update mode: " + updateInfo.getRowLevelUpdateMode());
         }
         Tuple2<RelNode, int[]> updateRelNodeAndRequireIndices =
                 convertToRowLevelUpdate(
@@ -647,7 +647,7 @@ private static void convertPredicateToNegative(LogicalTableModify tableModify) {
                     if (!(dynamicTableSource instanceof SupportsReadingMetadata)) {
                         throw new UnsupportedOperationException(
                                 String.format(
-                                        "The table source don't support reading metadata, but the require columns contains the meta columns: %s.",
+                                        "The table source does not support reading metadata, but the required columns contain metadata columns: %s.",
                                         column));
                     }
                     // list what metas the source supports to read
@@ -661,7 +661,7 @@ private static void convertPredicateToNegative(LogicalTableModify tableModify) {
                     if (!readableMetadata.containsKey(metaCol)) {
                         throw new IllegalArgumentException(
                                 String.format(
-                                        "Expect to read the meta column %s, but the table source for table %s doesn't support read the metadata column."
+                                        "Expected to read metadata column %s, but the table source for table %s does not support it. "
                                                 + "Please make sure the readable metadata for the source contains %s.",
                                         column,
                                         UnresolvedIdentifier.of(
@@ -674,7 +674,7 @@ private static void convertPredicateToNegative(LogicalTableModify tableModify) {
                     if (!dataType.equals(column.getDataType())) {
                         throw new IllegalArgumentException(
                                 String.format(
-                                        "Un-matched data type: the required column %s has datatype %s, but the data type in readable metadata for the table %s has data type %s. ",
+                                        "Mismatched data type: the required column %s has data type %s, but readable metadata for table %s has data type %s.",
                                         column,
                                         column.getDataType(),
                                         UnresolvedIdentifier.of(

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/spec/IntervalJoinSpec.java

@@ -28,9 +28,9 @@
 /**
  * IntervalJoinSpec describes how two tables will be joined in interval join.
  *
- * <p>This class corresponds to {@link org.apache.calcite.rel.core.Join} rel node. the join
- * condition is splitted into two part: WindowBounds and JoinSpec: 1. WindowBounds contains the time
- * range condition. 2. JoinSpec contains rest of the join condition except windowBounds.
+ * <p>This class corresponds to {@link org.apache.calcite.rel.core.Join} rel node. The join
+ * condition is splitted into two parts: WindowBounds and JoinSpec: 1. WindowBounds contains the
+ * time range condition. 2. JoinSpec contains rest of the join condition except windowBounds.
  */
 @JsonIgnoreProperties(ignoreUnknown = true)
 public class IntervalJoinSpec {

flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/GlobalAggregateITCase.java

@@ -63,7 +63,7 @@ void testExceptionThrowingAggregateFunction() throws Exception {
     /**
      * Source Function that uses updateGlobalAggregate() functionality exposed via
      * StreamingRuntimeContext to validate communication with JobMaster and test both failure and
-     * sucess scenarios.
+     * success scenarios.
      */
     private static class TestSourceFunction extends RichSourceFunction<Integer> {