Loader consumer group takes a lot of time to load based on the size of the batch and based on how loaded the redshift cluster is.

What we have seen when the cluster is resource starved, 100+ topics are being loaded to redshift concurrently and the batch size being operated by the loader is big (lakhs of rows in one load). Then the Commit to Kafka does not happen and the commits keep getting reprocessed.

What does not work

1. Big batch (msgBuf)
2. Auto commit = false

What works

1. Smaller batch (msgBuf)
2. Auto commit = true (from false, default of loader)

The cause is still not known

        if len(msgBuf) > 0 {
		lastMessage := msgBuf[len(msgBuf)-1]
		b.session.MarkOffset(
			lastMessage.Topic,
			lastMessage.Partition,
			lastMessage.Offset+1,
			"",
		)

		if b.autoCommit == false {
			b.session.Commit()
			klog.V(2).Infof("%s, Committed (autoCommit=false)", lastMessage.Topic)
		}

		b.lastCommittedOffset = lastMessage.Offset

	} else {
		klog.Warningf("%s, markOffset not possible for empty batch", b.topic)
	}

The b.session.Commit() does not have any error to return even though it is synchronous call. So we are not really sure at present why this happens as both marking offset and commit happens but the commit does not.

We are upgrading the redshift cluster as a short term fix.

Long term: discuss and debug to find the root cause.

• Reddit thread on this https://www.reddit.com/r/apachekafka/comments/lw7kdz/kafka_manual_commit_problem/

• Sarama Issue
  IBM/sarama#1894
  IBM/sarama#1843
  IBM/sarama#1570 (comment)

https://aws.amazon.com/premiumsupport/knowledge-center/redshift-query-abort/#:~:text=If%20your%20query%20is%20aborted,check%20the%20user%2Ddefined%20queues%3A&text=After%20you%20get%20the%20queue,defined%20in%20the%20WLM%20configuration.

d);, err: pq: Query (28572825) cancelled on user's request

SIGTERM is not leading to pod shutdown.

I0218 06:30:54.409376       1 main.go:102] SIGTERM signal received
I0218 06:30:54.409400       1 main.go:120] Cancelling context to trigger graceful shutdown...
I0218 06:30:54.409418       1 main.go:126] Waiting the some routines to gracefully shutdown (some don't)
I0218 06:30:54.663943       1 consumer.go:118] ConsumeClaim ended for topic: XX partition: 0
W0218 06:30:54.669826       1 schemaregistry.go:153] Retrying. Error getting latest schema, topic:X not found.

Until recreated the rsk object.

Please find and fix the issue.

F0219 09:19:51.767352       1 load_processor.go:236] Deduplication failed, pq: could not open relation with OID 12490225
goroutine 954 [running]:
github.com/practo/klog/v2.stacks(0xc00000e001, 0xc001646000, 0x79, 0x217)

https://stackoverflow.com/questions/5030720/tracing-the-cause-of-could-not-open-relation-with-oid-error

happens sometimes, rare but true

In the inital sink since loader lag is not there and defaultLag is being returned which is -1 and -1 is returned when the topic is not getting any request and its consumer group disappear.
Details: https://stackoverflow.com/questions/66294200/kafka-consumer-group-lag-initial-and-low-throughout-case-problem
This needs to be fixed.

The batcher library function should signal back to the main thread to not wait more and shutdown.

Kubernetes is designed to restart stuffs if they fail, for them to recover. Though we need to look at why it failed but a failure which lies as failed is more of problem and k8s restart does help.

Strimzi's KafkaConnect does not implement any probes, here is a discussion. strimzi/strimzi-kafka-operator#3850

We should do something for k8s restart to take effect if it cannot be done from Strimzi. It is very much required

groupID: ts-redshiftsink-latest-inventory-9f6b2f group ID is at present same in the operator for the batcher and loader consumers. This is not a problem since both operate on different topics?

Should we keep it separate? or same is an issue?

This needs to be debugged and fixed. Till this is fixed every loader process for table should be different.

F0911 09:42:15.528699 23410 load_processor.go:473] Error creating schema, err: pq: deadlock detected

mysql schema

 `Info` longtext,

debezium schema for the same thing

{
     "name": "Info",
      "type": [
          "null",
          "string"
       ],
      "default": null
},

When this data is loaded in redshift it fails as it expects the data type to be large. VARCHAR(MAX) but it is getting VARCHAR(255) since debezium is not transforming longtext to long.

Stackoverflow: https://stackoverflow.com/questions/63880976/debezium-mysql-longtext-to-debezium-data-type-conversions-is-not-correct

Every topic runs in its own go routine, so a Fatal in one routine impacts all the other go routines working on the data.
This is a very bad behaviour for the loader, as loader takes a lot of Redshift resources and issue in one of the loads of table impacts all the tables due to the shutdown due to the fatal. We should error and not fatal.

Fatal was for the development and stability phase and is no longer required. We should migrate to errors and retry particularly for the loader

Build Kubernetes CRDs and wrap the complete solution around.

kubectl get datapipeline

kubectl get redshifsink

https://twitter.com/alsingh87/status/1301892718000394241?s=20

Current Requirements

• Resource Management: Manage batcher and loader Kubernetes Deployment with all the configurations. CRD should wrap the batcher and loader deployment and manage sink by treating both as one sink group.
• Controller vs Resource: Separate out the controller part of configurations for a clean CRD resource specification. Like Redshift conf moves out of Loader conf into controller.
• [Mask Reloader] Parallel sink for mask configuration updates: Create two sink group, master and reloader, master always runs the current sink with current mask configuration. And reloader sink group sinks with a table suffix parallely to update the mask configuration in the tables.
• [Mask Reloader] Switch Sink Group: When reloader offsets consumed becomes close to the master offset a switch should happen
• Scaling batcher and loader: Need to scale memory of batcher based on number of entires being processed. When a lot of topics are present in batcher and all perform at high RPM then batcher runs out of memory. Need to solve this.
• GRANTs Handle redshift grants.

Merge needs to be performed only when the batch has a combination of update and delete operations. Otherwise skip merge and directly COPY to target table.

Performance improvement. First load are slow because of this not required merge.

Possible cause 1:
Mutilple table release one per reconcile is slow since it sinks all the sink groups again based on the state change. Treat multiple topic release as one activity and save time

Possible cause2:
fetch to kafka is not happening due to https://github.com/practo/tipoca-stream/issues

F0119 06:58:03.527475       1 batch_processor.go:326] Transforming schema:8 => inputTable failed: DebeziumType: record, SourceType: geometry, not handled

Need to handle conversion for geometry in mysql.

Stackoverflow 1

StackOverflow 2

Docs https://docs.aws.amazon.com/redshift/latest/dg/r_ALTER_TABLE_examples_basic.html

Find a way around this.

Works:

type changes

ALTER TABLE "datapipe.inventory"."customers" ALTER COLUMN first_name TYPE VARCHAR(300);

Does not work:

not null not supported

ALTER TABLE "datapipe.inventory"."customers" ALTER COLUMN first_name SET NOT NULL;
ERROR:  ALTER COLUMN SET NOT NULL is not supported

set and drop are not supported

ALTER TABLE "datapipe.inventory"."customers" ALTER COLUMN first_name SET DEFAULT "alok";
ERROR:  ALTER COLUMN SET/DROP DEFAULT is not supported

Prometheus metrics in each phase of the pipe.

At present there is a single resource requirement for all sinkgroups. But it should be configurable for all sink groups.

Please do not retry in the first attempt, after 10 retry failures things proceed. Retry is not needed in first case.

W0120 12:59:39.182145       1 schemaregistry.go:153] Retrying. Error getting latest schema, topic:loader-06e960-ts.inventory.customers, err:404 Not Found: Subject 'loader-06e960-ts.inventory.customers-value' not found.

Device a better algorithm to prevent repeated shuffling of sink groups without impacting the rollout time. Rollout time gets impacted due to releases happening only at 30mins interval (default).

When Schema Subject is not found, loader continues instead of failing.

It should crash or error.

• OffsetOldest stands for the oldest offset available on the broker for a partition. (-2)

• Initital: The initial offset to use if no offset was previously committed. Should be OffsetNewest or OffsetOldest. Defaults to OffsetNewest. Loader has it configured as OffsetOldest

Initial Offset for many consumer group claims started showing as -2(OffsetOldest). i.e. no data found, nothing was ever committed for these topics. But last committed offset for the topic does show nos > 0 but not for the same consumer group topic.

A drop in topics count was seen after a date. This looks related to Kafka data loss issue, but investigation is required.
Could be related to rebalancing as described here. IBM/sarama#1181

Either topic refresh is not working as expected, or the new topics are not getting picked up the loader consumer.

Keep the onea which are faster for small and bulk insert use cases..
https://stackoverflow.com/questions/63942859/unload-with-data-containing-quotes-and-delimiters?noredirect=1#comment113084781_63942859

This is a bug, fix it please

This bug has come since the types are not matching.

The type used to construct target table (GetTableMetadata) != inputTable

Guess is debezium timestamp date or datetime, one is still not handled as expected.

raw_line        | "XX","XX","XX","XX","2020-06-13 06:49:48","XX","","","2020-06-13 06:49:49","2020-06-13 06:50:05"

err_code        | 1206
err_reason      | Invalid timestamp format or value [YYYY-MM-DD HH24:MI:SS]

colname         | received_at

Support to run Grants on newly created tables for groups.

Helps the users of the database, with table replacements, grants changes get lost also should be done for the first time.

Batcher at present is not considering schema ID.

Batcher should consider it.

As data for a schema id needs to be operated together.

If ConsumeClaim exists, manager relaunches the batcher and it leads to masker getting created again and re pull of mask config.

F1223 05:47:10.258633       1 load_processor.go:561] Schema migration failed, err: 1 error occurred:
	* mismatch col: settlement_plan, prop: Name, input: settlement_plan, target: reach_plan_enabled

When column mismatch happens the migration is failing instead of applying the table reload. Big migrations like these should be handled by the operator. #37

Many tombstones messages for some topic.

I1013 11:38:35.411761       1 batcher.go:60] Skipping message, received a tombstone event}

redshiftbatcher/ ts.consult.doctor_languages/version1

2 diff rsk objects write to the same s3 object.

A new mask config change needs to refresh the complete table

S3 batcher does shutdown gracefully but every start should consider the last uploaded offset of batcher n the loader topic last message.

This will ensure regardless of what goes wrong, the system never looses data, never reprocess same data and thus becomes Idempotent.

Idea credits: Yelp experience https://engineeringblog.yelp.com/2016/10/redshift-connector.html

Intermiitently seeing this

I0224 12:11:26.424636       1 redshift.go:805] Running: UNLOAD from table_staged to s3
F0224 12:11:27.279014       1 load_processor.go:418] Error dropping staging table: cmd failed, cmd:DROP TABLE "schema"."table_adx_reload
_staged";, err: pq: Table "table_staged" does not exist

Feature to start streaming from a bin log postion.

When two rows are present for the same Kafka offset then deduplication does not happen.

practowarehouse=# select * from "inventory"."duplicate_staged";
-[ RECORD 1 ]----------------------
kafkaoffset | 1
operation   | CREATE
id          | [email protected]
-[ RECORD 2 ]----------------------
kafkaoffset | 2
operation   | CREATE
id          | [email protected]
-[ RECORD 3 ]----------------------
kafkaoffset | 1
operation   | CREATE
id          | [email protected]

delete from "inventory"."duplicate_staged" where kafkaoffset in (
select t1.kafkaoffset from "inventory"."duplicate_staged" t1 join "inventory"."duplicate_staged" t2 on t1.id=t2.id where t1.kafkaoffset < t2.kafkaoffset);

The above query wont deduplicate the data, since duplication is there for same kafka offset.

Failure due to too many parallel schema registry calls.

W0220 11:09:28.775278       1 schemaregistry.go:127] Retrying. Error fetching schema by id: 15 err:Get "https://schemareg.example.com/schemas/ids/15": context deadline exceeded (Client.Timeout exceeded while awaiting headers

Add cache.

Client already has cache. Please check that first.

This is only happening when 100s of topic are operated by the batcher at once. It is masking all data.

It is mostly due to mask config read not happening as exepected and the error is not crashing the batcher also.

Postgress shows/Redshift stl_load_errors: shows as this since all fields are masked.
err_reason | Invalid timestamp format or value [YYYY-MM-DD HH24:MI:SS]

/bug

During the first time sink of a database topic in Redshift-sink, the batcher gets 25k+ requests per second. Batcher is a memory batch which keeps the messages in memory before flushing the messages to the batch processor.

This leads to very high memory requirement for the batcher in the first time sink.

Need to work on this problem and solve it and make batcher work at very high speed keeping the memory footprint low.

https://github.com/practo/tipoca-stream/search?q=%2F%2F+TODO&type=Code

Getting broker not connected at the time of release

I0218 16:09:13.335796       1 sink_group_controller.go:615] ts.XX.X: waiting to reach realtime
E0218 16:09:14.030355       1 controller.go:267] controller-runtime/manager/controller/redshiftsink "msg"="Reconciler error" "error"="Failed to reconcile: kafka: broker not connected" "name"="XX" "namespace"="ts-redshiftsink-latest" "reconciler group"="tipoca.k8s.practo.dev" "reconciler kind"="RedshiftSink"

herryg91/gobatch#2

API server has errors for installing redshiftsink crd. It is due to the pod template generated CRD.

E0115 14:09:54.584777       1 customresource_handler.go:652] error building openapi models for redshiftsinks.tipoca.k8s.practo.dev: ERROR $root.definitions.dev.practo.k8s.tipoca.v1.RedshiftSink.properties.spec.properties.batcher.properties.podTemplate.properties.resources.properties.limits.additionalProperties.schema has invalid property: anyOf