diff --git a/models/collector_state.go b/models/collector_state.go
new file mode 100644
index 0000000..e0230a4
--- /dev/null
+++ b/models/collector_state.go
@@ -0,0 +1,35 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package models
+
+import (
+	"time"
+)
+
+type CollectorLatestState struct {
+	CreatedAt          time.Time `json:"createdAt"`
+	UpdatedAt          time.Time `json:"updatedAt"`
+	RawDataParams      string    `gorm:"primaryKey;column:raw_data_params;type:varchar(255);index" json:"raw_data_params"`
+	RawDataTable       string    `gorm:"primaryKey;column:raw_data_table;type:varchar(255)" json:"raw_data_table"`
+	CreatedDateAfter   *time.Time
+	LatestSuccessStart *time.Time
+}
+
+func (CollectorLatestState) TableName() string {
+	return "_devlake_collector_latest_state"
+}
diff --git a/models/migrationscripts/20221125_add_collector_state.go b/models/migrationscripts/20221125_add_collector_state.go
new file mode 100644
index 0000000..8aed5fc
--- /dev/null
+++ b/models/migrationscripts/20221125_add_collector_state.go
@@ -0,0 +1,39 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package migrationscripts
+
+import (
+	"github.com/apache/incubator-devlake/errors"
+	"github.com/apache/incubator-devlake/helpers/migrationhelper"
+	"github.com/apache/incubator-devlake/models/migrationscripts/archived"
+	"github.com/apache/incubator-devlake/plugins/core"
+)
+
+type addCollectorMeta20221125 struct{}
+
+func (script *addCollectorMeta20221125) Up(basicRes core.BasicRes) errors.Error {
+	return migrationhelper.AutoMigrateTables(basicRes, &archived.CollectorLatestState{})
+}
+
+func (*addCollectorMeta20221125) Version() uint64 {
+	return 20221125000038
+}
+
+func (*addCollectorMeta20221125) Name() string {
+	return "save state for collector"
+}
diff --git a/models/migrationscripts/archived/collector_state.go b/models/migrationscripts/archived/collector_state.go
new file mode 100644
index 0000000..e710803
--- /dev/null
+++ b/models/migrationscripts/archived/collector_state.go
@@ -0,0 +1,35 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package archived
+
+import (
+	"time"
+)
+
+type CollectorLatestState struct {
+	CreatedAt          time.Time `json:"createdAt"`
+	UpdatedAt          time.Time `json:"updatedAt"`
+	RawDataParams      string    `gorm:"primaryKey;column:raw_data_params;type:varchar(255);index" json:"raw_data_params"`
+	RawDataTable       string    `gorm:"primaryKey;column:raw_data_table;type:varchar(255)" json:"raw_data_table"`
+	CreatedDateAfter   *time.Time
+	LatestSuccessStart *time.Time
+}
+
+func (CollectorLatestState) TableName() string {
+	return "_devlake_collector_latest_state"
+}
diff --git a/models/migrationscripts/register.go b/models/migrationscripts/register.go
index f9d1029..ad39df3 100644
--- a/models/migrationscripts/register.go
+++ b/models/migrationscripts/register.go
@@ -63,5 +63,6 @@
 		new(addLabels),
 		new(renameFiledsInProjectPrMetric),
 		new(addEnableToProjectMetric),
+		new(addCollectorMeta20221125),
 	}
 }
diff --git a/plugins/core/plugin_task.go b/plugins/core/plugin_task.go
index ea09e98..5716f5b 100644
--- a/plugins/core/plugin_task.go
+++ b/plugins/core/plugin_task.go
@@ -104,7 +104,7 @@
 	PrepareTaskData(taskCtx TaskContext, options map[string]interface{}) (interface{}, errors.Error)
 }
 
-// CloseablePluginTask Extends PluginTask, and invokes a Close method after all subtasks are done
+// CloseablePluginTask Extends PluginTask, and invokes a Close method after all subtasks are done or fail
 type CloseablePluginTask interface {
 	PluginTask
 	Close(taskCtx TaskContext) errors.Error
diff --git a/plugins/helper/api_collector_with_state.go b/plugins/helper/api_collector_with_state.go
new file mode 100644
index 0000000..6404b75
--- /dev/null
+++ b/plugins/helper/api_collector_with_state.go
@@ -0,0 +1,94 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package helper
+
+import (
+	"github.com/apache/incubator-devlake/errors"
+	"github.com/apache/incubator-devlake/models"
+	"github.com/apache/incubator-devlake/plugins/core/dal"
+	"gorm.io/gorm"
+	"time"
+)
+
+// ApiCollectorStateManager save collector state in framework table
+type ApiCollectorStateManager struct {
+	RawDataSubTaskArgs
+	*ApiCollector
+	LatestState      models.CollectorLatestState
+	CreatedDateAfter *time.Time
+	ExecuteStart     time.Time
+}
+
+// NewApiCollectorWithState create a new ApiCollectorStateManager
+func NewApiCollectorWithState(args RawDataSubTaskArgs, createdDateAfter *time.Time) (*ApiCollectorStateManager, errors.Error) {
+	db := args.Ctx.GetDal()
+
+	rawDataSubTask, err := NewRawDataSubTask(args)
+	if err != nil {
+		return nil, errors.Default.Wrap(err, "Couldn't resolve raw subtask args")
+	}
+	latestState := models.CollectorLatestState{}
+	err = db.First(&latestState, dal.Where(`raw_data_table = ? AND raw_data_params = ?`, rawDataSubTask.table, rawDataSubTask.params))
+	if err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			latestState = models.CollectorLatestState{
+				RawDataTable:  rawDataSubTask.table,
+				RawDataParams: rawDataSubTask.params,
+			}
+		} else {
+			return nil, errors.Default.Wrap(err, "failed to load JiraLatestCollectorMeta")
+		}
+	}
+	return &ApiCollectorStateManager{
+		RawDataSubTaskArgs: args,
+		LatestState:        latestState,
+		CreatedDateAfter:   createdDateAfter,
+		ExecuteStart:       time.Now(),
+	}, nil
+}
+
+// CanIncrementCollect return if the old data can support collect incrementally.
+// only when latest collection is success &&
+// (m.LatestState.CreatedDateAfter == nil means all data have been collected ||
+// CreatedDateAfter at this time exists and later than in the LatestState)
+// if CreatedDateAfter at this time not exists, collect incrementally only when "m.LatestState.CreatedDateAfter == nil"
+func (m ApiCollectorStateManager) CanIncrementCollect() bool {
+	return m.LatestState.LatestSuccessStart != nil &&
+		(m.LatestState.CreatedDateAfter == nil || m.CreatedDateAfter != nil && m.CreatedDateAfter.After(*m.LatestState.CreatedDateAfter))
+}
+
+// InitCollector init the embedded collector
+func (m *ApiCollectorStateManager) InitCollector(args ApiCollectorArgs) (err errors.Error) {
+	args.RawDataSubTaskArgs = m.RawDataSubTaskArgs
+	m.ApiCollector, err = NewApiCollector(args)
+	return err
+}
+
+// Execute the embedded collector and record execute state
+func (m ApiCollectorStateManager) Execute() errors.Error {
+	err := m.ApiCollector.Execute()
+	if err != nil {
+		return err
+	}
+
+	db := m.Ctx.GetDal()
+	m.LatestState.LatestSuccessStart = &m.ExecuteStart
+	m.LatestState.CreatedDateAfter = m.CreatedDateAfter
+	err = db.CreateOrUpdate(&m.LatestState)
+	return err
+}
diff --git a/plugins/jira/impl/impl.go b/plugins/jira/impl/impl.go
index 29aaf36..2aff632 100644
--- a/plugins/jira/impl/impl.go
+++ b/plugins/jira/impl/impl.go
@@ -171,9 +171,9 @@
 		return nil, errors.Default.Wrap(err, "unable to get Jira connection")
 	}
 
-	var since time.Time
-	if op.Since != "" {
-		since, err = time.Parse("2006-01-02T15:04:05Z", op.Since)
+	var createdDateAfter time.Time
+	if op.CreatedDateAfter != "" {
+		createdDateAfter, err = time.Parse("2006-01-02T15:04:05Z", op.CreatedDateAfter)
 		if err != nil {
 			return nil, errors.BadInput.Wrap(err, "invalid value for `since`")
 		}
@@ -208,10 +208,11 @@
 		ApiClient:      jiraApiClient,
 		JiraServerInfo: *info,
 	}
-	if !since.IsZero() {
-		taskData.Since = &since
-		logger.Debug("collect data updated since %s", since)
+	if !createdDateAfter.IsZero() {
+		taskData.CreatedDateAfter = &createdDateAfter
+		logger.Debug("collect data created from %s", createdDateAfter)
 	}
+
 	return taskData, nil
 }
 
diff --git a/plugins/jira/jira.go b/plugins/jira/jira.go
index 0cf85de..dcb50c3 100644
--- a/plugins/jira/jira.go
+++ b/plugins/jira/jira.go
@@ -32,12 +32,12 @@
 	boardId := cmd.Flags().Uint64P("board", "b", 0, "jira board id")
 	_ = cmd.MarkFlagRequired("connection")
 	_ = cmd.MarkFlagRequired("board")
-	since := cmd.Flags().StringP("since", "s", "", "collect data that are updated after specified time, ie 2006-05-06T07:08:09Z")
+	CreatedDateAfter := cmd.Flags().StringP("created_date_after", "a", "", "collect data that are updated after specified time, ie 2006-05-06T07:08:09Z")
 	cmd.Run = func(c *cobra.Command, args []string) {
 		runner.DirectRun(c, args, PluginEntry, map[string]interface{}{
-			"connectionId": *connectionId,
-			"boardId":      *boardId,
-			"since":        *since,
+			"connectionId":     *connectionId,
+			"boardId":          *boardId,
+			"CreatedDateAfter": *CreatedDateAfter,
 		})
 	}
 	runner.RunCmd(cmd)
diff --git a/plugins/jira/tasks/epic_collector.go b/plugins/jira/tasks/epic_collector.go
index 78e6c31..ef4c270 100644
--- a/plugins/jira/tasks/epic_collector.go
+++ b/plugins/jira/tasks/epic_collector.go
@@ -53,12 +53,7 @@
 	if err != nil {
 		return err
 	}
-	since := data.Since
 	jql := "ORDER BY created ASC"
-	if since != nil {
-		// prepend a time range criteria if `since` was specified, either by user or from database
-		jql = fmt.Sprintf("updated >= '%s' %s", since.Format("2006/01/02 15:04"), jql)
-	}
 	collector, err := helper.NewApiCollector(helper.ApiCollectorArgs{
 		RawDataSubTaskArgs: helper.RawDataSubTaskArgs{
 			Ctx: taskCtx,
diff --git a/plugins/jira/tasks/issue_changelog_collector.go b/plugins/jira/tasks/issue_changelog_collector.go
index dc3e6f1..30c47cd 100644
--- a/plugins/jira/tasks/issue_changelog_collector.go
+++ b/plugins/jira/tasks/issue_changelog_collector.go
@@ -52,6 +52,18 @@
 	log := taskCtx.GetLogger()
 	db := taskCtx.GetDal()
 
+	collectorWithState, err := helper.NewApiCollectorWithState(helper.RawDataSubTaskArgs{
+		Ctx: taskCtx,
+		Params: JiraApiParams{
+			ConnectionId: data.Options.ConnectionId,
+			BoardId:      data.Options.BoardId,
+		},
+		Table: RAW_CHANGELOG_TABLE,
+	}, data.CreatedDateAfter)
+	if err != nil {
+		return err
+	}
+
 	// query for issue_ids that needed changelog collection
 	clauses := []dal.Clause{
 		dal.Select("i.issue_id, i.updated AS update_time"),
@@ -60,12 +72,10 @@
 		dal.Join("LEFT JOIN _tool_jira_issue_changelogs c ON (c.connection_id = i.connection_id AND c.issue_id = i.issue_id)"),
 		dal.Where("i.updated > i.created AND bi.connection_id = ?  AND bi.board_id = ? AND i.std_type != ? ", data.Options.ConnectionId, data.Options.BoardId, "Epic"),
 		dal.Groupby("i.issue_id, i.updated"),
-		dal.Having("i.updated > max(c.issue_updated) OR  (max(c.issue_updated) IS NULL AND COUNT(c.changelog_id) > 0)"),
 	}
-	// apply time range if any
-	since := data.Since
-	if since != nil {
-		clauses = append(clauses, dal.Where("i.updated > ?", *since))
+	incremental := collectorWithState.CanIncrementCollect()
+	if incremental {
+		clauses = append(clauses, dal.Having("i.updated > max(c.issue_updated) OR  (max(c.issue_updated) IS NULL AND COUNT(c.changelog_id) > 0)"))
 	}
 
 	if log.IsLevelEnabled(core.LOG_DEBUG) {
@@ -88,18 +98,10 @@
 	}
 
 	// now, let ApiCollector takes care the rest
-	collector, err := helper.NewApiCollector(helper.ApiCollectorArgs{
-		RawDataSubTaskArgs: helper.RawDataSubTaskArgs{
-			Ctx: taskCtx,
-			Params: JiraApiParams{
-				ConnectionId: data.Options.ConnectionId,
-				BoardId:      data.Options.BoardId,
-			},
-			Table: RAW_CHANGELOG_TABLE,
-		},
+	err = collectorWithState.InitCollector(helper.ApiCollectorArgs{
 		ApiClient:     data.ApiClient,
 		PageSize:      100,
-		Incremental:   since == nil,
+		Incremental:   incremental,
 		GetTotalPages: GetTotalPagesFromResponse,
 		Input:         iterator,
 		UrlTemplate:   "api/3/issue/{{ .Input.IssueId }}/changelog",
@@ -127,5 +129,5 @@
 		return err
 	}
 
-	return collector.Execute()
+	return collectorWithState.Execute()
 }
diff --git a/plugins/jira/tasks/issue_collector.go b/plugins/jira/tasks/issue_collector.go
index 50db842..2120e6b 100644
--- a/plugins/jira/tasks/issue_collector.go
+++ b/plugins/jira/tasks/issue_collector.go
@@ -21,26 +21,20 @@
 	"encoding/json"
 	goerror "errors"
 	"fmt"
-	"github.com/apache/incubator-devlake/errors"
+	"gorm.io/gorm"
 	"io"
 	"net/http"
 	"net/url"
 
+	"github.com/apache/incubator-devlake/errors"
 	"github.com/apache/incubator-devlake/plugins/core"
 	"github.com/apache/incubator-devlake/plugins/core/dal"
 	"github.com/apache/incubator-devlake/plugins/helper"
 	"github.com/apache/incubator-devlake/plugins/jira/models"
-	"gorm.io/gorm"
 )
 
 const RAW_ISSUE_TABLE = "jira_api_issues"
 
-// this struct should be moved to `jira_api_common.go`
-type JiraApiParams struct {
-	ConnectionId uint64
-	BoardId      uint64
-}
-
 var _ core.SubTaskEntryPoint = CollectIssues
 
 var CollectIssuesMeta = core.SubTaskMeta{
@@ -55,10 +49,38 @@
 	db := taskCtx.GetDal()
 	data := taskCtx.GetData().(*JiraTaskData)
 
-	since := data.Since
-	incremental := false
-	// user didn't specify a time range to sync, try load from database
-	if since == nil {
+	collectorWithState, err := helper.NewApiCollectorWithState(helper.RawDataSubTaskArgs{
+		Ctx: taskCtx,
+		/*
+			This struct will be JSONEncoded and stored into database along with raw data itself, to identity minimal
+			set of data to be process, for example, we process JiraIssues by Board
+		*/
+		Params: JiraApiParams{
+			ConnectionId: data.Options.ConnectionId,
+			BoardId:      data.Options.BoardId,
+		},
+		/*
+			Table store raw data
+		*/
+		Table: RAW_ISSUE_TABLE,
+	}, data.CreatedDateAfter)
+	if err != nil {
+		return err
+	}
+
+	// build jql
+	// IMPORTANT: we have to keep paginated data in a consistence order to avoid data-missing, if we sort issues by
+	//  `updated`, issue will be jumping between pages if it got updated during the collection process
+	createdDateAfter := data.CreatedDateAfter
+	jql := "ORDER BY created ASC"
+	if createdDateAfter != nil {
+		// prepend a time range criteria if `since` was specified, either by user or from database
+		jql = fmt.Sprintf("created >= '%v' %v", createdDateAfter.Format("2006/01/02 15:04"), jql)
+	}
+
+	incremental := collectorWithState.CanIncrementCollect()
+	if incremental {
+		// user didn't specify a time range to sync, try load from database
 		var latestUpdated models.JiraIssue
 		clauses := []dal.Clause{
 			dal.Select("_tool_jira_issues.*"),
@@ -72,35 +94,13 @@
 			return errors.NotFound.Wrap(err, "failed to get latest jira issue record")
 		}
 		if latestUpdated.IssueId > 0 {
-			since = &latestUpdated.Updated
-			incremental = true
+			jql = fmt.Sprintf("updated >= '%v' %v", latestUpdated.Updated.Format("2006/01/02 15:04"), jql)
+		} else {
+			incremental = false
 		}
 	}
-	// build jql
-	// IMPORTANT: we have to keep paginated data in a consistence order to avoid data-missing, if we sort issues by
-	//  `updated`, issue will be jumping between pages if it got updated during the collection process
-	jql := "ORDER BY created ASC"
-	if since != nil {
-		// prepend a time range criteria if `since` was specified, either by user or from database
-		jql = fmt.Sprintf("updated >= '%v' %v", since.Format("2006/01/02 15:04"), jql)
-	}
 
-	collector, err := helper.NewApiCollector(helper.ApiCollectorArgs{
-		RawDataSubTaskArgs: helper.RawDataSubTaskArgs{
-			Ctx: taskCtx,
-			/*
-				This struct will be JSONEncoded and stored into database along with raw data itself, to identity minimal
-				set of data to be process, for example, we process JiraIssues by Board
-			*/
-			Params: JiraApiParams{
-				ConnectionId: data.Options.ConnectionId,
-				BoardId:      data.Options.BoardId,
-			},
-			/*
-				Table store raw data
-			*/
-			Table: RAW_ISSUE_TABLE,
-		},
+	err = collectorWithState.InitCollector(helper.ApiCollectorArgs{
 		ApiClient:   data.ApiClient,
 		PageSize:    100,
 		Incremental: incremental,
@@ -159,10 +159,9 @@
 			return data.Issues, nil
 		},
 	})
-
 	if err != nil {
 		return err
 	}
 
-	return collector.Execute()
+	return collectorWithState.Execute()
 }
diff --git a/plugins/jira/tasks/issue_extractor.go b/plugins/jira/tasks/issue_extractor.go
index bf6f623..7f61c52 100644
--- a/plugins/jira/tasks/issue_extractor.go
+++ b/plugins/jira/tasks/issue_extractor.go
@@ -111,7 +111,7 @@
 	if issue.ResolutionDate != nil {
 		issue.LeadTimeMinutes = uint(issue.ResolutionDate.Unix()-issue.Created.Unix()) / 60
 	}
-	if data.Options.TransformationRules.StoryPointField != "" {
+	if data.Options.TransformationRules != nil && data.Options.TransformationRules.StoryPointField != "" {
 		unknownStoryPoint := apiIssue.Fields.AllFields[data.Options.TransformationRules.StoryPointField]
 		switch sp := unknownStoryPoint.(type) {
 		case string:
@@ -193,9 +193,11 @@
 	}
 	stdTypeMappings := make(map[string]string)
 	standardStatusMappings := make(map[string]StatusMappings)
-	for userType, stdType := range data.Options.TransformationRules.TypeMappings {
-		stdTypeMappings[userType] = strings.ToUpper(stdType.StandardType)
-		standardStatusMappings[userType] = stdType.StatusMappings
+	if data.Options.TransformationRules != nil {
+		for userType, stdType := range data.Options.TransformationRules.TypeMappings {
+			stdTypeMappings[userType] = strings.ToUpper(stdType.StandardType)
+			standardStatusMappings[userType] = stdType.StatusMappings
+		}
 	}
 	return &typeMappings{
 		typeIdMappings:         typeIdMapping,
diff --git a/plugins/jira/tasks/remotelink_collector.go b/plugins/jira/tasks/remotelink_collector.go
index b641f9d..4867225 100644
--- a/plugins/jira/tasks/remotelink_collector.go
+++ b/plugins/jira/tasks/remotelink_collector.go
@@ -47,6 +47,18 @@
 	logger := taskCtx.GetLogger()
 	logger.Info("collect remotelink")
 
+	collectorWithState, err := helper.NewApiCollectorWithState(helper.RawDataSubTaskArgs{
+		Ctx: taskCtx,
+		Params: JiraApiParams{
+			ConnectionId: data.Options.ConnectionId,
+			BoardId:      data.Options.BoardId,
+		},
+		Table: RAW_REMOTELINK_TABLE,
+	}, data.CreatedDateAfter)
+	if err != nil {
+		return err
+	}
+
 	clauses := []dal.Clause{
 		dal.Select("i.issue_id, i.updated AS update_time"),
 		dal.From("_tool_jira_board_issues bi"),
@@ -54,12 +66,14 @@
 		dal.Join("LEFT JOIN _tool_jira_remotelinks rl ON (rl.connection_id = i.connection_id AND rl.issue_id = i.issue_id)"),
 		dal.Where("i.updated > i.created AND bi.connection_id = ?  AND bi.board_id = ?  ", data.Options.ConnectionId, data.Options.BoardId),
 		dal.Groupby("i.issue_id, i.updated"),
-		dal.Having("i.updated > max(rl.issue_updated) OR  max(rl.issue_updated) IS NULL"),
 	}
-	// apply time range if any
-	since := data.Since
-	if since != nil {
-		clauses = append(clauses, dal.Where("i.updated > ?", *since))
+	incremental := collectorWithState.CanIncrementCollect()
+	if incremental {
+		if collectorWithState.LatestState.LatestSuccessStart != nil {
+			clauses = append(clauses, dal.Having("i.updated > ? AND (i.updated > max(rl.issue_updated) OR max(rl.issue_updated) IS NULL)", collectorWithState.LatestState.LatestSuccessStart))
+		} else {
+			clauses = append(clauses, dal.Having("i.updated > max(rl.issue_updated) OR max(rl.issue_updated) IS NULL"))
+		}
 	}
 	cursor, err := db.Cursor(clauses...)
 	if err != nil {
@@ -73,18 +87,10 @@
 		return err
 	}
 
-	collector, err := helper.NewApiCollector(helper.ApiCollectorArgs{
-		RawDataSubTaskArgs: helper.RawDataSubTaskArgs{
-			Ctx: taskCtx,
-			Params: JiraApiParams{
-				ConnectionId: data.Options.ConnectionId,
-				BoardId:      data.Options.BoardId,
-			},
-			Table: RAW_REMOTELINK_TABLE,
-		},
+	err = collectorWithState.InitCollector(helper.ApiCollectorArgs{
 		ApiClient:   data.ApiClient,
 		Input:       iterator,
-		Incremental: since == nil,
+		Incremental: incremental,
 		UrlTemplate: "api/2/issue/{{ .Input.IssueId }}/remotelink",
 		ResponseParser: func(res *http.Response) ([]json.RawMessage, errors.Error) {
 			if res.StatusCode == http.StatusNotFound {
@@ -102,7 +108,7 @@
 	if err != nil {
 		return err
 	}
-	err = collector.Execute()
+	err = collectorWithState.Execute()
 	if err != nil {
 		return err
 	}
diff --git a/plugins/jira/tasks/remotelink_extractor.go b/plugins/jira/tasks/remotelink_extractor.go
index b857e26..0418a1a 100644
--- a/plugins/jira/tasks/remotelink_extractor.go
+++ b/plugins/jira/tasks/remotelink_extractor.go
@@ -45,7 +45,8 @@
 	logger.Info("extract remote links")
 	var commitShaRegex *regexp.Regexp
 	var err error
-	if pattern := data.Options.TransformationRules.RemotelinkCommitShaPattern; pattern != "" {
+	if data.Options.TransformationRules != nil && data.Options.TransformationRules.RemotelinkCommitShaPattern != "" {
+		pattern := data.Options.TransformationRules.RemotelinkCommitShaPattern
 		commitShaRegex, err = regexp.Compile(pattern)
 		if err != nil {
 			return errors.Default.Wrap(err, "regexp Compile pattern failed")
diff --git a/plugins/jira/tasks/task_data.go b/plugins/jira/tasks/task_data.go
index b4bf770..1e99812 100644
--- a/plugins/jira/tasks/task_data.go
+++ b/plugins/jira/tasks/task_data.go
@@ -94,17 +94,22 @@
 type JiraOptions struct {
 	ConnectionId         uint64 `json:"connectionId"`
 	BoardId              uint64 `json:"boardId"`
-	Since                string
+	CreatedDateAfter     string
 	TransformationRules  *JiraTransformationRule `json:"transformationRules"`
 	ScopeId              string
 	TransformationRuleId uint64
 }
 
 type JiraTaskData struct {
-	Options        *JiraOptions
-	ApiClient      *helper.ApiAsyncClient
-	Since          *time.Time
-	JiraServerInfo models.JiraServerInfo
+	Options          *JiraOptions
+	ApiClient        *helper.ApiAsyncClient
+	CreatedDateAfter *time.Time
+	JiraServerInfo   models.JiraServerInfo
+}
+
+type JiraApiParams struct {
+	ConnectionId uint64
+	BoardId      uint64
 }
 
 func DecodeAndValidateTaskOptions(options map[string]interface{}) (*JiraOptions, errors.Error) {
diff --git a/plugins/jira/tasks/worklog_collector.go b/plugins/jira/tasks/worklog_collector.go
index 0c3ce7a..3958a81 100644
--- a/plugins/jira/tasks/worklog_collector.go
+++ b/plugins/jira/tasks/worklog_collector.go
@@ -42,10 +42,21 @@
 func CollectWorklogs(taskCtx core.SubTaskContext) errors.Error {
 	db := taskCtx.GetDal()
 	data := taskCtx.GetData().(*JiraTaskData)
-	since := data.Since
 
 	logger := taskCtx.GetLogger()
 
+	collectorWithState, err := helper.NewApiCollectorWithState(helper.RawDataSubTaskArgs{
+		Ctx: taskCtx,
+		Params: JiraApiParams{
+			ConnectionId: data.Options.ConnectionId,
+			BoardId:      data.Options.BoardId,
+		},
+		Table: RAW_WORKLOGS_TABLE,
+	}, data.CreatedDateAfter)
+	if err != nil {
+		return err
+	}
+
 	// filter out issue_ids that needed collection
 	clauses := []dal.Clause{
 		dal.Select("i.issue_id, i.updated AS update_time"),
@@ -54,11 +65,10 @@
 		dal.Join("LEFT JOIN _tool_jira_worklogs wl ON (wl.connection_id = i.connection_id AND wl.issue_id = i.issue_id)"),
 		dal.Where("i.updated > i.created AND bi.connection_id = ?  AND bi.board_id = ?  ", data.Options.ConnectionId, data.Options.BoardId),
 		dal.Groupby("i.issue_id, i.updated"),
-		dal.Having("i.updated > max(wl.issue_updated) OR  (max(wl.issue_updated) IS NULL AND COUNT(wl.worklog_id) > 0)"),
 	}
-	// apply time range if any
-	if since != nil {
-		clauses = append(clauses, dal.Where("i.updated > ?", *since))
+	incremental := collectorWithState.CanIncrementCollect()
+	if incremental {
+		clauses = append(clauses, dal.Having("i.updated > max(wl.issue_updated) OR  (max(wl.issue_updated) IS NULL AND COUNT(wl.worklog_id) > 0)"))
 	}
 
 	// construct the input iterator
@@ -71,20 +81,12 @@
 		return err
 	}
 
-	collector, err := helper.NewApiCollector(helper.ApiCollectorArgs{
-		RawDataSubTaskArgs: helper.RawDataSubTaskArgs{
-			Ctx: taskCtx,
-			Params: JiraApiParams{
-				ConnectionId: data.Options.ConnectionId,
-				BoardId:      data.Options.BoardId,
-			},
-			Table: RAW_WORKLOGS_TABLE,
-		},
+	err = collectorWithState.InitCollector(helper.ApiCollectorArgs{
 		Input:         iterator,
 		ApiClient:     data.ApiClient,
 		UrlTemplate:   "api/2/issue/{{ .Input.IssueId }}/worklog",
 		PageSize:      50,
-		Incremental:   since == nil,
+		Incremental:   incremental,
 		GetTotalPages: GetTotalPagesFromResponse,
 		ResponseParser: func(res *http.Response) ([]json.RawMessage, errors.Error) {
 			var data struct {
@@ -103,5 +105,5 @@
 		return err
 	}
 
-	return collector.Execute()
+	return collectorWithState.Execute()
 }