@ -33,41 +33,36 @@ import org.apache.flink.cdc.common.schema.Column;
import org.apache.flink.cdc.common.schema.Schema ;
import org.apache.flink.cdc.common.sink.MetadataApplier ;
import org.apache.flink.cdc.common.types.DataType ;
import org.apache.flink.cdc.runtime.operators.schema.event.RefreshPendingListsResponse ;
import org.apache.flink.cdc.runtime.operators.schema.event.ReleaseUpstreamRequest ;
import org.apache.flink.cdc.runtime.operators.schema.event.ReleaseUpstreamResponse ;
import org.apache.flink.cdc.common.utils.Preconditions ;
import org.apache.flink.cdc.runtime.operators.schema.event.SchemaChangeProcessingResponse ;
import org.apache.flink.cdc.runtime.operators.schema.event.SchemaChangeRequest ;
import org.apache.flink.cdc.runtime.operators.schema.event.SchemaChangeResponse ;
import org.apache.flink.cdc.runtime.operators.schema.event.SchemaChangeResultResponse ;
import org.apache.flink.runtime.operators.coordination.CoordinationResponse ;
import org.slf4j.Logger ;
import org.slf4j.LoggerFactory ;
import javax.annotation.concurrent.NotThreadSafe ;
import java.io.Closeable ;
import java.io.IOException ;
import java.util.ArrayList ;
import java.util.Collections ;
import java.util.HashMap ;
import java.util.HashSet ;
import java.util.LinkedList ;
import java.util.List ;
import java.util.Map ;
import java.util.Set ;
import java.util.concurrent.CompletableFuture ;
import java.util.concurrent.ExecutorService ;
import java.util.concurrent.Executors ;
import java.util.concurrent.atomic.AtomicReference ;
import java.util.stream.Collectors ;
import java.util.stream.Stream ;
import static org.apache.flink.cdc.runtime.operators.schema.coordinator.SchemaRegistryRequestHandler.RequestStatus.RECEIVED_RELEASE_REQUEST ;
import static org.apache.flink.cdc.runtime.operators.schema.event.CoordinationResponseUtils.wrap ;
/** A handler to deal with all requests and events for {@link SchemaRegistry}. */
@Internal
@NotThreadSafe
public class SchemaRegistryRequestHandler implements Closeable {
private static final Logger LOG = LoggerFactory . getLogger ( SchemaRegistryRequestHandler . class ) ;
@ -81,21 +76,18 @@ public class SchemaRegistryRequestHandler implements Closeable {
private final SchemaDerivation schemaDerivation ;
/ * *
* Not applied SchemaChangeRequest before receiving all flush success events for its table from
* sink writers .
* Atomic flag indicating if current RequestHandler could accept more schema changes for now .
* /
private final List< PendingSchemaChange > pendingSchemaChange s;
private final AtomicReference< RequestStatus > schemaChangeStatu s;
private final List < SchemaChangeEvent > finishedSchemaChanges ;
private final List < SchemaChangeEvent > ignoredSchemaChanges ;
private volatile Throwable currentChangeException ;
private volatile List < SchemaChangeEvent > currentDerivedSchemaChangeEvents ;
private volatile List < SchemaChangeEvent > currentFinishedSchemaChanges ;
private volatile List < SchemaChangeEvent > currentIgnoredSchemaChanges ;
/** Sink writers which have sent flush success events for the request. */
private final Set < Integer > flushedSinkWriters ;
/** Status of the execution of current schema change request. */
private volatile boolean isSchemaChangeApplying ;
/** Actual exception if failed to apply schema change. */
private volatile Throwable schemaChangeException ;
/** Executor service to execute schema change. */
private final ExecutorService schemaChangeThreadPool ;
@ -107,16 +99,81 @@ public class SchemaRegistryRequestHandler implements Closeable {
SchemaDerivation schemaDerivation ,
SchemaChangeBehavior schemaChangeBehavior ) {
this . metadataApplier = metadataApplier ;
this . activeSinkWriters = new HashSet < > ( ) ;
this . flushedSinkWriters = new HashSet < > ( ) ;
this . pendingSchemaChanges = new LinkedList < > ( ) ;
this . finishedSchemaChanges = new LinkedList < > ( ) ;
this . ignoredSchemaChanges = new LinkedList < > ( ) ;
this . schemaManager = schemaManager ;
this . schemaDerivation = schemaDerivation ;
this . schemaChangeThreadPool = Executors . newSingleThreadExecutor ( ) ;
this . isSchemaChangeApplying = false ;
this . schemaChangeBehavior = schemaChangeBehavior ;
this . activeSinkWriters = new HashSet < > ( ) ;
this . flushedSinkWriters = new HashSet < > ( ) ;
this . schemaChangeThreadPool = Executors . newSingleThreadExecutor ( ) ;
this . currentDerivedSchemaChangeEvents = new ArrayList < > ( ) ;
this . currentFinishedSchemaChanges = new ArrayList < > ( ) ;
this . currentIgnoredSchemaChanges = new ArrayList < > ( ) ;
this . schemaChangeStatus = new AtomicReference < > ( RequestStatus . IDLE ) ;
}
/ * *
* Handle the { @link SchemaChangeRequest } and wait for all sink subtasks flushing .
*
* @param request the received SchemaChangeRequest
* /
public CompletableFuture < CoordinationResponse > handleSchemaChangeRequest (
SchemaChangeRequest request ) {
if ( schemaChangeStatus . compareAndSet ( RequestStatus . IDLE , RequestStatus . WAITING_FOR_FLUSH ) ) {
LOG . info (
"Received schema change event request {} from table {}. Start to buffer requests for others." ,
request . getSchemaChangeEvent ( ) ,
request . getTableId ( ) . toString ( ) ) ;
SchemaChangeEvent event = request . getSchemaChangeEvent ( ) ;
// If this schema change event has been requested by another subTask, ignore it.
if ( schemaManager . isOriginalSchemaChangeEventRedundant ( event ) ) {
LOG . info ( "Event {} has been addressed before, ignoring it." , event ) ;
clearCurrentSchemaChangeRequest ( ) ;
Preconditions . checkState (
schemaChangeStatus . compareAndSet (
RequestStatus . WAITING_FOR_FLUSH , RequestStatus . IDLE ) ,
"Illegal schemaChangeStatus state: should still in WAITING_FOR_FLUSH state if event was duplicated." ) ;
return CompletableFuture . completedFuture ( wrap ( SchemaChangeResponse . duplicate ( ) ) ) ;
}
schemaManager . applyOriginalSchemaChange ( event ) ;
List < SchemaChangeEvent > derivedSchemaChangeEvents =
calculateDerivedSchemaChangeEvents ( request . getSchemaChangeEvent ( ) ) ;
// If this schema change event is filtered out by LENIENT mode or merging table route
// strategies, ignore it.
if ( derivedSchemaChangeEvents . isEmpty ( ) ) {
LOG . info ( "Event {} is omitted from sending to downstream, ignoring it." , event ) ;
clearCurrentSchemaChangeRequest ( ) ;
Preconditions . checkState (
schemaChangeStatus . compareAndSet (
RequestStatus . WAITING_FOR_FLUSH , RequestStatus . IDLE ) ,
"Illegal schemaChangeStatus state: should still in WAITING_FOR_FLUSH state if event was ignored." ) ;
return CompletableFuture . completedFuture ( wrap ( SchemaChangeResponse . ignored ( ) ) ) ;
}
// Backfill pre-schema info for sink applying
derivedSchemaChangeEvents . forEach (
e - > {
if ( e instanceof SchemaChangeEventWithPreSchema ) {
SchemaChangeEventWithPreSchema pe = ( SchemaChangeEventWithPreSchema ) e ;
if ( ! pe . hasPreSchema ( ) ) {
schemaManager
. getLatestEvolvedSchema ( pe . tableId ( ) )
. ifPresent ( pe : : fillPreSchema ) ;
}
}
} ) ;
currentDerivedSchemaChangeEvents = new ArrayList < > ( derivedSchemaChangeEvents ) ;
return CompletableFuture . completedFuture (
wrap ( SchemaChangeResponse . accepted ( derivedSchemaChangeEvents ) ) ) ;
} else {
LOG . info (
"Schema Registry is busy processing a schema change request, could not handle request {} for now." ,
request ) ;
return CompletableFuture . completedFuture ( wrap ( SchemaChangeResponse . busy ( ) ) ) ;
}
}
/ * *
@ -127,27 +184,22 @@ public class SchemaRegistryRequestHandler implements Closeable {
* /
private void applySchemaChange (
TableId tableId , List < SchemaChangeEvent > derivedSchemaChangeEvents ) {
isSchemaChangeApplying = true ;
schemaChangeException = null ;
finishedSchemaChanges . clear ( ) ;
ignoredSchemaChanges . clear ( ) ;
for ( SchemaChangeEvent changeEvent : derivedSchemaChangeEvents ) {
if ( changeEvent . getType ( ) ! = SchemaChangeEventType . CREATE_TABLE ) {
if ( schemaChangeBehavior = = SchemaChangeBehavior . IGNORE ) {
i gnoredSchemaChanges. add ( changeEvent ) ;
currentIgnoredSchemaChanges . add ( changeEvent ) ;
continue ;
}
}
if ( ! metadataApplier . acceptsSchemaEvolutionType ( changeEvent . getType ( ) ) ) {
LOG . info ( "Ignored schema change {} to table {}." , changeEvent , tableId ) ;
i gnoredSchemaChanges. add ( changeEvent ) ;
currentI gnoredSchemaChanges. add ( changeEvent ) ;
} else {
try {
metadataApplier . applySchemaChange ( changeEvent ) ;
LOG . debug ( "Applied schema change {} to table {}." , changeEvent , tableId ) ;
LOG . info ( "Applied schema change {} to table {}." , changeEvent , tableId ) ;
schemaManager . applyEvolvedSchemaChange ( changeEvent ) ;
f inishedSchemaChanges. add ( changeEvent ) ;
currentF inishedSchemaChanges. add ( changeEvent ) ;
} catch ( Throwable t ) {
LOG . error (
"Failed to apply schema change {} to table {}. Caused by: {}" ,
@ -155,7 +207,7 @@ public class SchemaRegistryRequestHandler implements Closeable {
tableId ,
t ) ;
if ( ! shouldIgnoreException ( t ) ) {
schema ChangeException = t ;
current ChangeException = t ;
break ;
} else {
LOG . warn (
@ -166,74 +218,9 @@ public class SchemaRegistryRequestHandler implements Closeable {
}
}
}
PendingSchemaChange waitFlushSuccess = pendingSchemaChanges . get ( 0 ) ;
if ( RECEIVED_RELEASE_REQUEST . equals ( waitFlushSuccess . getStatus ( ) ) ) {
startNextSchemaChangeRequest ( ) ;
}
isSchemaChangeApplying = false ;
}
/ * *
* Handle the { @link SchemaChangeRequest } and wait for all sink subtasks flushing .
*
* @param request the received SchemaChangeRequest
* /
public CompletableFuture < CoordinationResponse > handleSchemaChangeRequest (
SchemaChangeRequest request ) {
if ( pendingSchemaChanges . isEmpty ( ) ) {
LOG . info (
"Received schema change event request from table {}. Start to buffer requests for others." ,
request . getTableId ( ) . toString ( ) ) ;
SchemaChangeEvent event = request . getSchemaChangeEvent ( ) ;
if ( event instanceof CreateTableEvent
& & schemaManager . originalSchemaExists ( request . getTableId ( ) ) ) {
return CompletableFuture . completedFuture (
wrap ( new SchemaChangeResponse ( Collections . emptyList ( ) ) ) ) ;
}
schemaManager . applyOriginalSchemaChange ( event ) ;
List < SchemaChangeEvent > derivedSchemaChangeEvents =
calculateDerivedSchemaChangeEvents ( request . getSchemaChangeEvent ( ) ) ;
derivedSchemaChangeEvents . forEach (
e - > {
if ( e instanceof SchemaChangeEventWithPreSchema ) {
SchemaChangeEventWithPreSchema pe = ( SchemaChangeEventWithPreSchema ) e ;
if ( ! pe . hasPreSchema ( ) ) {
schemaManager
. getLatestEvolvedSchema ( pe . tableId ( ) )
. ifPresent ( pe : : fillPreSchema ) ;
}
}
} ) ;
CompletableFuture < CoordinationResponse > response =
CompletableFuture . completedFuture (
wrap ( new SchemaChangeResponse ( derivedSchemaChangeEvents ) ) ) ;
if ( ! derivedSchemaChangeEvents . isEmpty ( ) ) {
PendingSchemaChange pendingSchemaChange =
new PendingSchemaChange ( request , response ) ;
pendingSchemaChange . derivedSchemaChangeEvents = derivedSchemaChangeEvents ;
pendingSchemaChanges . add ( pendingSchemaChange ) ;
pendingSchemaChanges . get ( 0 ) . startToWaitForReleaseRequest ( ) ;
}
return response ;
} else {
LOG . info ( "There are already processing requests. Wait for processing." ) ;
CompletableFuture < CoordinationResponse > response = new CompletableFuture < > ( ) ;
pendingSchemaChanges . add ( new PendingSchemaChange ( request , response ) ) ;
return response ;
}
}
/** Handle the {@link ReleaseUpstreamRequest} and wait for all sink subtasks flushing. */
public CompletableFuture < CoordinationResponse > handleReleaseUpstreamRequest ( ) {
CompletableFuture < CoordinationResponse > response =
pendingSchemaChanges . get ( 0 ) . getResponseFuture ( ) ;
if ( response . isDone ( ) & & ! isSchemaChangeApplying ) {
startNextSchemaChangeRequest ( ) ;
} else {
pendingSchemaChanges . get ( 0 ) . receiveReleaseRequest ( ) ;
}
return response ;
Preconditions . checkState (
schemaChangeStatus . compareAndSet ( RequestStatus . APPLYING , RequestStatus . FINISHED ) ,
"Illegal schemaChangeStatus state: should be APPLYING before applySchemaChange finishes" ) ;
}
/ * *
@ -252,76 +239,33 @@ public class SchemaRegistryRequestHandler implements Closeable {
* @param tableId the subtask in SchemaOperator and table that the FlushEvent is about
* @param sinkSubtask the sink subtask succeed flushing
* /
public void flushSuccess ( TableId tableId , int sinkSubtask ) throws InterruptedException {
public void flushSuccess ( TableId tableId , int sinkSubtask ) {
flushedSinkWriters . add ( sinkSubtask ) ;
if ( flushedSinkWriters . equals ( activeSinkWriters ) ) {
Preconditions . checkState (
schemaChangeStatus . compareAndSet (
RequestStatus . WAITING_FOR_FLUSH , RequestStatus . APPLYING ) ,
"Illegal schemaChangeStatus state: should be WAITING_FOR_FLUSH before collecting enough FlushEvents" ) ;
LOG . info (
"All sink subtask have flushed for table {}. Start to apply schema change." ,
tableId . toString ( ) ) ;
PendingSchemaChange waitFlushSuccess = pendingSchemaChanges . get ( 0 ) ;
schemaChangeThreadPool . submit (
( ) - > applySchemaChange ( tableId , waitFlushSuccess . derivedSchemaChangeEvents ) ) ;
Thread . sleep ( 1000 ) ;
if ( schemaChangeException ! = null ) {
throw new RuntimeException ( "Failed to apply schema change." , schemaChangeException ) ;
}
if ( isSchemaChangeApplying ) {
waitFlushSuccess
. getResponseFuture ( )
. complete ( wrap ( new SchemaChangeProcessingResponse ( ) ) ) ;
} else {
waitFlushSuccess
. getResponseFuture ( )
. complete ( wrap ( new ReleaseUpstreamResponse ( finishedSchemaChanges ) ) ) ;
}
( ) - > applySchemaChange ( tableId , currentDerivedSchemaChangeEvents ) ) ;
}
}
private void startNextSchemaChangeRequest ( ) {
pendingSchemaChanges . remove ( 0 ) ;
flushedSinkWriters . clear ( ) ;
while ( ! pendingSchemaChanges . isEmpty ( ) ) {
PendingSchemaChange pendingSchemaChange = pendingSchemaChanges . get ( 0 ) ;
SchemaChangeRequest request = pendingSchemaChange . changeRequest ;
if ( request . getSchemaChangeEvent ( ) instanceof CreateTableEvent
& & schemaManager . evolvedSchemaExists ( request . getTableId ( ) ) ) {
pendingSchemaChange
. getResponseFuture ( )
. complete ( wrap ( new SchemaChangeResponse ( Collections . emptyList ( ) ) ) ) ;
pendingSchemaChanges . remove ( 0 ) ;
} else {
List < SchemaChangeEvent > derivedSchemaChangeEvents =
calculateDerivedSchemaChangeEvents ( request . getSchemaChangeEvent ( ) ) ;
pendingSchemaChange
. getResponseFuture ( )
. complete ( wrap ( new SchemaChangeResponse ( derivedSchemaChangeEvents ) ) ) ;
if ( ! derivedSchemaChangeEvents . isEmpty ( ) ) {
pendingSchemaChange . derivedSchemaChangeEvents = derivedSchemaChangeEvents ;
pendingSchemaChange . startToWaitForReleaseRequest ( ) ;
break ;
}
}
}
}
public CompletableFuture < CoordinationResponse > refreshPendingLists ( ) {
pendingSchemaChanges . clear ( ) ;
flushedSinkWriters . clear ( ) ;
return CompletableFuture . completedFuture ( wrap ( new RefreshPendingListsResponse ( ) ) ) ;
}
public CompletableFuture < CoordinationResponse > getSchemaChangeResult ( ) {
if ( schemaChangeException ! = null ) {
throw new RuntimeException ( "Failed to apply schema change." , schemaChangeException ) ;
}
if ( isSchemaChangeApplying ) {
return CompletableFuture . supplyAsync ( ( ) - > wrap ( new SchemaChangeProcessingResponse ( ) ) ) ;
} else {
Preconditions . checkState (
! schemaChangeStatus . get ( ) . equals ( RequestStatus . IDLE ) ,
"Illegal schemaChangeStatus: should not be IDLE before getting schema change request results." ) ;
if ( schemaChangeStatus . compareAndSet ( RequestStatus . FINISHED , RequestStatus . IDLE ) ) {
// This request has been finished, return it and prepare for the next request
List < SchemaChangeEvent > finishedEvents = clearCurrentSchemaChangeRequest ( ) ;
return CompletableFuture . supplyAsync (
( ) - > wrap ( new ReleaseUpstreamResponse ( finishedSchemaChanges ) ) ) ;
( ) - > wrap ( new SchemaChangeResultResponse ( finishedEvents ) ) ) ;
} else {
// Still working on schema change request, waiting it
return CompletableFuture . supplyAsync ( ( ) - > wrap ( new SchemaChangeProcessingResponse ( ) ) ) ;
}
}
@ -445,57 +389,54 @@ public class SchemaRegistryRequestHandler implements Closeable {
}
private boolean shouldIgnoreException ( Throwable throwable ) {
// In IGNORE mode, will never try to apply schema change events
// In EVOLVE and and LENIENT mode, such failure will not be tolerated
// In EVOLVE and LENIENT mode, such failure will not be tolerated
// In EXCEPTION mode, an exception will be thrown once captured
return ( throwable instanceof UnsupportedSchemaChangeEventException )
& & ( schemaChangeBehavior = = SchemaChangeBehavior . TRY_EVOLVE ) ;
}
private static class PendingSchemaChange {
private final SchemaChangeRequest changeRequest ;
private List < SchemaChangeEvent > derivedSchemaChangeEvents ;
private CompletableFuture < CoordinationResponse > responseFuture ;
private RequestStatus status ;
public PendingSchemaChange (
SchemaChangeRequest changeRequest ,
CompletableFuture < CoordinationResponse > responseFuture ) {
this . changeRequest = changeRequest ;
this . responseFuture = responseFuture ;
this . status = RequestStatus . PENDING ;
}
public SchemaChangeRequest getChangeRequest ( ) {
return changeRequest ;
}
public CompletableFuture < CoordinationResponse > getResponseFuture ( ) {
return responseFuture ;
}
public RequestStatus getStatus ( ) {
return status ;
}
public void startToWaitForReleaseRequest ( ) {
if ( ! responseFuture . isDone ( ) ) {
throw new IllegalStateException (
"Cannot start to wait for flush success before the SchemaChangeRequest is done." ) ;
}
this . responseFuture = new CompletableFuture < > ( ) ;
this . status = RequestStatus . WAIT_RELEASE_REQUEST ;
}
public void receiveReleaseRequest ( ) {
this . status = RECEIVED_RELEASE_REQUEST ;
private List < SchemaChangeEvent > clearCurrentSchemaChangeRequest ( ) {
if ( currentChangeException ! = null ) {
throw new RuntimeException ( "Failed to apply schema change." , currentChangeException ) ;
}
List < SchemaChangeEvent > finishedSchemaChanges =
new ArrayList < > ( currentFinishedSchemaChanges ) ;
flushedSinkWriters . clear ( ) ;
currentDerivedSchemaChangeEvents . clear ( ) ;
currentFinishedSchemaChanges . clear ( ) ;
currentIgnoredSchemaChanges . clear ( ) ;
currentChangeException = null ;
return finishedSchemaChanges ;
}
enum RequestStatus {
PENDING ,
WAIT_RELEASE_REQUEST ,
RECEIVED_RELEASE_REQUEST
// Schema change event state could transfer in the following way:
//
// -------- B --------
// | |
// v |
// -------- ---------------------
// | IDLE | --- A --> | WAITING_FOR_FLUSH |
// -------- ---------------------
// ^ |
// E C
// \ v
// ------------ ------------
// | FINISHED | <-- D -- | APPLYING |
// ------------ ------------
//
// A: When a request came to an idling request handler.
// B: When current request is duplicate or ignored by LENIENT / routed table merging
// strategies.
// C: When schema registry collected enough flush success events, and actually started to apply
// schema changes.
// D: When schema change application finishes (successfully or with exceptions)
// E: When current schema change request result has been retrieved by SchemaOperator, and ready
// for the next request.
private enum RequestStatus {
IDLE ,
WAITING_FOR_FLUSH ,
APPLYING ,
FINISHED
}
}
yuxiqian
5 months ago
committed by
GitHub