bulker: disable consumers for stale topics

jitsucom · Jan 15, 2024 · 3a409df · 3a409df
1 parent dc7a847
commit 3a409df
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 2 deletions.
diff --git a/bulkerapp/app/integration_test.go b/bulkerapp/app/integration_test.go
@@ -105,7 +105,8 @@ func initApp(t *testing.T, envVars map[string]string) (app *appbase.App[Config],
 // Test streams in autocommit and bath mode. Both with good batches and batches with primary key violation error
 func TestGoodAndBadStreams(t *testing.T) {
 	app, kafkaContainer, postgresContainer := initApp(t, map[string]string{"BULKER_MESSAGES_RETRY_COUNT": "0",
-		"BULKER_BATCH_RUNNER_DEFAULT_PERIOD_SEC": "1"})
+		"BULKER_TOPIC_MANAGER_REFRESH_PERIOD_SEC": "1",
+		"BULKER_BATCH_RUNNER_DEFAULT_PERIOD_SEC":  "1"})
 	t.Cleanup(func() {
 		app.Exit(appbase.SIG_SHUTDOWN_FOR_TESTS)
 		time.Sleep(5 * time.Second)
@@ -183,6 +184,7 @@ func TestEventsRetry(t *testing.T) {
 		"BULKER_BATCH_RUNNER_DEFAULT_RETRY_PERIOD_SEC":     "5",
 		"BULKER_BATCH_RUNNER_DEFAULT_RETRY_BATCH_FRACTION": "1",
 		"BULKER_MESSAGES_RETRY_BACKOFF_BASE":               "0",
+		"BULKER_TOPIC_MANAGER_REFRESH_PERIOD_SEC":          "1",
 		"BULKER_BATCH_RUNNER_DEFAULT_PERIOD_SEC":           "1"})
 	t.Cleanup(func() {
 		app.Exit(appbase.SIG_SHUTDOWN_FOR_TESTS)

diff --git a/bulkerapp/app/topic_manager.go b/bulkerapp/app/topic_manager.go
@@ -148,7 +148,7 @@ func (tm *TopicManager) LoadMetadata() {
 		tm.Errorf("Error getting topic offsets: %v", err)
 	} else {
 		for tp, offset := range res.ResultInfos {
-			if offset.Offset > 0 {
+			if offset.Offset >= 0 {
 				topicsLastMessageDates[*tp.Topic] = time.UnixMilli(offset.Timestamp)
 			} else {
 				topicsLastMessageDates[*tp.Topic] = time.Time{}