[Improvement][new api] refer to apache#2127

1. class name of SeaTunnelRuntimeEnvironment changes to SeaTunnelContextAware
2. Serializer doesn't need to extend Serializable
3. unify Enumerator concept

seatunnel-api/src/main/java/org/apache/seatunnel/api/serialization/Serializer.java

@@ -18,9 +18,8 @@
 package org.apache.seatunnel.api.serialization;
 
 import java.io.IOException;
-import java.io.Serializable;
 
-public interface Serializer<T> extends Serializable {
+public interface Serializer<T> {
 
     /**
      * Serializes the given object.

seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/SeaTunnelSink.java

@@ -20,7 +20,7 @@
 import org.apache.seatunnel.api.common.PluginIdentifierInterface;
 import org.apache.seatunnel.api.common.SeaTunnelPluginLifeCycle;
 import org.apache.seatunnel.api.serialization.Serializer;
-import org.apache.seatunnel.api.source.SeaTunnelRuntimeEnvironment;
+import org.apache.seatunnel.api.source.SeaTunnelContextAware;
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 
@@ -43,7 +43,7 @@
  *                                {@link SinkAggregatedCommitter} handle it, this class should implement interface {@link Serializable}.
  */
 public interface SeaTunnelSink<IN, StateT, CommitInfoT, AggregatedCommitInfoT>
-    extends Serializable, PluginIdentifierInterface, SeaTunnelPluginLifeCycle, SeaTunnelRuntimeEnvironment {
+    extends Serializable, PluginIdentifierInterface, SeaTunnelPluginLifeCycle, SeaTunnelContextAware {
 
     /**
      * Set the row type info of sink row data. This method will be automatically called by translation.

seatunnel-api/src/main/java/org/apache/seatunnel/api/source/SeaTunnelRuntimeEnvironment.java → seatunnel-api/src/main/java/org/apache/seatunnel/api/source/SeaTunnelContextAware.java

@@ -22,7 +22,7 @@
 /**
  * This interface defines the runtime environment of the SeaTunnel application.
  */
-public interface SeaTunnelRuntimeEnvironment {
+public interface SeaTunnelContextAware {
 
     default void setSeaTunnelContext(SeaTunnelContext seaTunnelContext){
         // nothing

seatunnel-api/src/main/java/org/apache/seatunnel/api/source/SeaTunnelSource.java

@@ -34,7 +34,7 @@
  * @param <StateT> The type of checkpoint states.
  */
 public interface SeaTunnelSource<T, SplitT extends SourceSplit, StateT>
-    extends Serializable, PluginIdentifierInterface, SeaTunnelPluginLifeCycle, SeaTunnelRuntimeEnvironment {
+    extends Serializable, PluginIdentifierInterface, SeaTunnelPluginLifeCycle, SeaTunnelContextAware {
 
     /**
      * Get the boundedness of this source.

seatunnel-api/src/main/java/org/apache/seatunnel/api/source/SourceReader.java

@@ -114,6 +114,6 @@ interface Context {
          *
          * @param sourceEvent the source event to coordinator.
          */
-        void sendSourceEventToCoordinator(SourceEvent sourceEvent);
+        void sendSourceEventToEnumerator(SourceEvent sourceEvent);
     }
 }

seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/source/AbstractSingleSplitSource.java

@@ -29,7 +29,7 @@ public abstract class AbstractSingleSplitSource<T> implements SeaTunnelSource<T,
 
     @Override
     public final AbstractSingleSplitReader<T> createReader(SourceReader.Context readerContext) throws Exception {
-        checkArgument(readerContext.getIndexOfSubtask() == 0, "Single split source allows only a single reader to be created.");
+        checkArgument(readerContext.getIndexOfSubtask() == 0, "A single split source allows only one single reader to be created.");
         return createReader(new SingleSplitReaderContext(readerContext));
     }
 

seatunnel-translation/seatunnel-translation-base/src/main/java/org/apache/seatunnel/translation/source/CoordinatedReaderContext.java

@@ -56,7 +56,7 @@ public void sendSplitRequest() {
     }
 
     @Override
-    public void sendSourceEventToCoordinator(SourceEvent sourceEvent) {
+    public void sendSourceEventToEnumerator(SourceEvent sourceEvent) {
         coordinatedSource.handleReaderEvent(subtaskId, sourceEvent);
     }
 }

seatunnel-translation/seatunnel-translation-base/src/main/java/org/apache/seatunnel/translation/source/ParallelReaderContext.java

@@ -56,7 +56,7 @@ public void sendSplitRequest() {
     }
 
     @Override
-    public void sendSourceEventToCoordinator(SourceEvent sourceEvent) {
+    public void sendSourceEventToEnumerator(SourceEvent sourceEvent) {
         // TODO: exception
         throw new RuntimeException("");
     }