Hi all, I am trying flink table store ,which i want to be an product solution. but there is same trouble : Here is my main code:

// ... decode input as POJO Datastream
 SingleOutputStreamOperator<PushLogRow> rows=input.process(new InputProcessFunction());

// ... convert to table
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
Table inputTable = tableEnv.fromDataStream(rows);
inputTable.printSchema();
// write to table store
FlinkCatalog catalog= FlinkCatalogFactory.createCatalog("flink-table",tableCfg);
tableEnv.registerCatalog("flink-table",catalog);
tableEnv.useCatalog("flink-table");
tableEnv.useDatabase("default");
inputTable.executeInsert("t_push_log");

the first problem is DataStream Column Type not match with TABLE on type of Map, then i find a way to resolved: add anotation to my POJO class:

@TypeInfo(FlinkTypeInfoFactory.class)
public class PushLogRow{

   private Map<String,String> ex;

	@JsonAnySetter
	public void setEx(String key, Object value){
		if (null==ex){
			ex =new HashMap<>();
		}
		ex.put(key,String.valueOf(value));
	}
}

public class FlinkTypeInfoFactory<T> extends TypeInfoFactory<T> {

    @Override
    public TypeInformation<T> createTypeInfo(Type t, Map<String, TypeInformation<?>> map) {
        Map<String, TypeInformation<?>> typeInfo = new HashMap<>();
        Class<T> clazz=(Class<T>) t;
        FieldUtils.getAllFieldsList(clazz).forEach(field -> {
            String key=field.getName();
            Class<?> type=field.getType();
             ...
            }else if (type==Map.class){
                typeInfo.put(key, Types.MAP(Types.STRING,Types.STRING));
            }
        });
        return Types.POJO(clazz,typeInfo);
    }
}

then the DataStream table schema printed as I want Map<String,String> instead of RAW(java.util.Map,...) but Could not perform checkpoint, which error as blow:

13:57:39,072 WARN  org.apache.flink.runtime.taskmanager.Task                    [] - Writer (3/8)#0 (a18ee02afc5d80bf846f1cfbc8863239_8d96fc510e75de3baf03ef7367db7d42_2_0) switched from RUNNING to FAILED with failure cause: java.io.IOException: Could not perform checkpoint 1 for operator Writer (3/8)#0.
	at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointOnBarrier(StreamTask.java:1238)
	at org.apache.flink.streaming.runtime.io.checkpointing.CheckpointBarrierHandler.notifyCheckpoint(CheckpointBarrierHandler.java:147)
	at org.apache.flink.streaming.runtime.io.checkpointing.SingleCheckpointBarrierHandler.triggerCheckpoint(SingleCheckpointBarrierHandler.java:287)
	at org.apache.flink.streaming.runtime.io.checkpointing.SingleCheckpointBarrierHandler.access$100(SingleCheckpointBarrierHandler.java:64)
	at org.apache.flink.streaming.runtime.io.checkpointing.SingleCheckpointBarrierHandler$ControllerImpl.triggerGlobalCheckpoint(SingleCheckpointBarrierHandler.java:488)
	at org.apache.flink.streaming.runtime.io.checkpointing.AbstractAlignedBarrierHandlerState.triggerGlobalCheckpoint(AbstractAlignedBarrierHandlerState.java:74)
	at org.apache.flink.streaming.runtime.io.checkpointing.AbstractAlignedBarrierHandlerState.barrierReceived(AbstractAlignedBarrierHandlerState.java:66)
	at org.apache.flink.streaming.runtime.io.checkpointing.SingleCheckpointBarrierHandler.lambda$processBarrier$2(SingleCheckpointBarrierHandler.java:234)
	at org.apache.flink.streaming.runtime.io.checkpointing.SingleCheckpointBarrierHandler.markCheckpointAlignedAndTransformState(SingleCheckpointBarrierHandler.java:262)
	at org.apache.flink.streaming.runtime.io.checkpointing.SingleCheckpointBarrierHandler.processBarrier(SingleCheckpointBarrierHandler.java:231)
	at org.apache.flink.streaming.runtime.io.checkpointing.CheckpointedInputGate.handleEvent(CheckpointedInputGate.java:181)
	at org.apache.flink.streaming.runtime.io.checkpointing.CheckpointedInputGate.pollNext(CheckpointedInputGate.java:159)
	at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:110)
	at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:542)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:831)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:780)
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:935)
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:914)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:728)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:550)
	at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: java.io.IOException: java.lang.IndexOutOfBoundsException
	at org.apache.flink.table.store.connector.sink.StoreWriteOperator.prepareCommit(StoreWriteOperator.java:192)
	at org.apache.flink.table.store.connector.sink.PrepareCommitOperator.emitCommittables(PrepareCommitOperator.java:59)
	at org.apache.flink.table.store.connector.sink.PrepareCommitOperator.prepareSnapshotPreBarrier(PrepareCommitOperator.java:47)
	at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.prepareSnapshotPreBarrier(RegularOperatorChain.java:89)
	at org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.checkpointState(SubtaskCheckpointCoordinatorImpl.java:334)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$performCheckpoint$13(StreamTask.java:1281)
	at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:1269)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointOnBarrier(StreamTask.java:1226)
	... 22 more
Caused by: java.lang.IndexOutOfBoundsException
	at org.apache.flink.core.memory.MemorySegment.get(MemorySegment.java:375)
	at org.apache.flink.table.data.binary.BinaryStringData.compareTo(BinaryStringData.java:178)
	at MemTableComparator$1775.compare(Unknown Source)
	at org.apache.flink.table.runtime.operators.sort.BinaryIndexedSortable.compareRecords(BinaryIndexedSortable.java:184)
	at org.apache.flink.table.runtime.operators.sort.BinaryIndexedSortable.compare(BinaryIndexedSortable.java:176)
	at org.apache.flink.runtime.operators.sort.QuickSort.sortInternal(QuickSort.java:136)
	at org.apache.flink.runtime.operators.sort.QuickSort.sortInternal(QuickSort.java:321)
	at org.apache.flink.runtime.operators.sort.QuickSort.sortInternal(QuickSort.java:315)
	at org.apache.flink.runtime.operators.sort.QuickSort.sortInternal(QuickSort.java:315)
	at org.apache.flink.runtime.operators.sort.QuickSort.sortInternal(QuickSort.java:321)
	at org.apache.flink.runtime.operators.sort.QuickSort.sortInternal(QuickSort.java:321)
	at org.apache.flink.runtime.operators.sort.QuickSort.sortInternal(QuickSort.java:315)
	at org.apache.flink.runtime.operators.sort.QuickSort.sort(QuickSort.java:66)
	at org.apache.flink.runtime.operators.sort.QuickSort.sort(QuickSort.java:81)
	at org.apache.flink.table.store.file.mergetree.SortBufferMemTable.mergeIterator(SortBufferMemTable.java:110)
	at org.apache.flink.table.store.file.mergetree.MergeTreeWriter.flushMemory(MergeTreeWriter.java:160)
	at org.apache.flink.table.store.file.mergetree.MergeTreeWriter.prepareCommit(MergeTreeWriter.java:184)
	at org.apache.flink.table.store.table.sink.AbstractTableWrite.prepareCommit(AbstractTableWrite.java:96)
	at org.apache.flink.table.store.connector.sink.StoreWriteOperator.prepareCommit(StoreWriteOperator.java:188)
13:57:39,072 WARN  org.apache.flink.runtime.taskmanager.Task                    [] - Writer (3/8)#0 (a18ee02afc5d80bf846f1cfbc8863239_8d96fc510e75de3baf03ef7367db7d42_2_0) switched from RUNNING to FAILED with failure cause: java.io.IOException: Could not perform checkpoint 1 for operator Writer (3/8)#0.
	at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointOnBarrier(StreamTask.java:1238)
	at org.apache.flink.streaming.runtime.io.checkpointing.CheckpointBarrierHandler.notifyCheckpoint(CheckpointBarrierHandler.java:147)
	at org.apache.flink.streaming.runtime.io.checkpointing.SingleCheckpointBarrierHandler.triggerCheckpoint(SingleCheckpointBarrierHandler.java:287)
	at org.apache.flink.streaming.runtime.io.checkpointing.SingleCheckpointBarrierHandler.access$100(SingleCheckpointBarrierHandler.java:64)
	at org.apache.flink.streaming.runtime.io.checkpointing.SingleCheckpointBarrierHandler$ControllerImpl.triggerGlobalCheckpoint(SingleCheckpointBarrierHandler.java:488)
	at org.apache.flink.streaming.runtime.io.checkpointing.AbstractAlignedBarrierHandlerState.triggerGlobalCheckpoint(AbstractAlignedBarrierHandlerState.java:74)
	at org.apache.flink.streaming.runtime.io.checkpointing.AbstractAlignedBarrierHandlerState.barrierReceived(AbstractAlignedBarrierHandlerState.java:66)
	at org.apache.flink.streaming.runtime.io.checkpointing.SingleCheckpointBarrierHandler.lambda$processBarrier$2(SingleCheckpointBarrierHandler.java:234)
	at org.apache.flink.streaming.runtime.io.checkpointing.SingleCheckpointBarrierHandler.markCheckpointAlignedAndTransformState(SingleCheckpointBarrierHandler.java:262)
    ...
Caused by: java.io.IOException: java.lang.IndexOutOfBoundsException
	at org.apache.flink.table.store.connector.sink.StoreWriteOperator.prepareCommit(StoreWriteOperator.java:192)
	at org.apache.flink.table.store.connector.sink.PrepareCommitOperator.emitCommittables(PrepareCommitOperator.java:59)
	at org.apache.flink.table.store.connector.sink.PrepareCommitOperator.prepareSnapshotPreBarrier(PrepareCommitOperator.java:47)
	at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.prepareSnapshotPreBarrier(RegularOperatorChain.java:89)
	at org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.checkpointState(SubtaskCheckpointCoordinatorImpl.java:334)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$performCheckpoint$13(StreamTask.java:1281)
	at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:1269)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointOnBarrier(StreamTask.java:1226)
	... 22 more
Caused by: java.lang.IndexOutOfBoundsException
	at org.apache.flink.core.memory.MemorySegment.get(MemorySegment.java:375)
	at org.apache.flink.table.data.binary.BinaryStringData.compareTo(BinaryStringData.java:178)
	at MemTableComparator$1775.compare(Unknown Source)
	at org.apache.flink.table.runtime.operators.sort.BinaryIndexedSortable.compareRecords(BinaryIndexedSortable.java:184)

when droped Map field working well my env: flink-1.16.0-bin-scala_2.12 with flink-table-store-dist-0.2.1.jar here is samething i want to be helped: 1. does my way is correct or effect, any better way? 2. is the problem is a bug , having other solution or not ? thank all for any concern reply, i will be extremely grateful.