[HUDI-1621] Gets the parallelism from context when init StreamWriteOperatorCoordinator

• Type: Improvement
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 0.8.0
• Component/s: flink
• Labels: pull-request-available
• Assignee: lamber-ken
• Reporter: Xie Lei
• Created: 16/Feb/21 10:21
• Updated: 17/Feb/21 12:05
• Resolved: 17/Feb/21 12:05

Description

If the parallelism in the constructor of StreamWriteOperatorFactory is not equal to the parallelism of the Operator, the flink task will run failed.

For example, if the StreamWriteOperatorFactory's parallelism less than Operator's parallelism

6807 [flink-akka.actor.default-dispatcher-4] INFO  org.apache.flink.runtime.jobmaster.JobMaster  - Trying to recover from a global failure.
java.lang.ArrayIndexOutOfBoundsException: 3
  at org.apache.hudi.operator.StreamWriteOperatorCoordinator.handleEventFromOperator(StreamWriteOperatorCoordinator.java:181)
  at org.apache.flink.runtime.operators.coordination.OperatorCoordinatorHolder.handleEventFromOperator(OperatorCoordinatorHolder.java:191)
  at org.apache.flink.runtime.scheduler.SchedulerBase.deliverOperatorEventToCoordinator(SchedulerBase.java:952)
  at org.apache.flink.runtime.jobmaster.JobMaster.sendOperatorEventToCoordinator(JobMaster.java:473)
  at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
  at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  at java.lang.reflect.Method.invoke(Method.java:498)
  at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:284)
  at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:199)
  at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:74)
  at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:152)
  at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26)
  at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21)

 

Issue Links

• links to: GitHub Pull Request #2579