fix: Registry long type (#34)

* Fix avsc schema options for long type * Update Debezium consumer debug formats
itcig · Sep 15, 2020 · 2df7a7f · 2df7a7f
1 parent e04ed2e
commit 2df7a7f
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 8 deletions.
diff --git a/src/application/registry/registry.ts b/src/application/registry/registry.ts
@@ -1,6 +1,5 @@
 import { SchemaRegistry } from '@kafkajs/confluent-schema-registry';
 import avro from 'avsc';
-import deepmerge from 'deepmerge';
 import { RegistryConfig, RegistryOptions } from '../../types/types';
 
 import longType from './types/long-type';
@@ -10,9 +9,8 @@ export class KafkaSchemaRegistry extends SchemaRegistry {
 		// // Merge any passed forSchemaOptions with custom registry for long Type.
 		// // Without specifying long Type then any BigInts from a database would be truncated and throw a precision error
 		// // https://github.com/kafkajs/confluent-schema-registry/issues/53
-		super(
-			schemaRegistryClient,
-			deepmerge(options || {}, {
+		super(schemaRegistryClient, {
+			forSchemaOptions: {
 				registry: {
 					long: longType,
 				},
@@ -31,7 +29,8 @@ export class KafkaSchemaRegistry extends SchemaRegistry {
 					// Return the type registered with the same name, if any.
 					return opts.registry[name];
 				},
-			})
-		);
+			},
+			...options,
+		});
 	}
 }
diff --git a/src/application/services/consumers/debezium-mysql/debezium-mysql.class.ts b/src/application/services/consumers/debezium-mysql/debezium-mysql.class.ts
@@ -68,7 +68,7 @@ export class DebeziumMysqlConsumerService extends ConsumerService {
 			eachBatch: async ({ batch, resolveOffset, commitOffsetsIfNecessary, heartbeat, isRunning, isStale }): Promise<void> => {
 				Debug('metamorphosis:app:consumer:debezium-mysql:debug')(`Consuming batch of ${batch.messages.length} messages`);
 
-				Debug('metamorphosis:app:consumer:debezium-mysql:debug')('Consuming batch: %O', {
+				Debug('metamorphosis:app:consumer:debezium-mysql:debug')('Consuming batch: %o', {
 					topic: batch.topic,
 					partition: batch.partition,
 					highWatermark: batch.highWatermark,
@@ -79,7 +79,7 @@ export class DebeziumMysqlConsumerService extends ConsumerService {
 					if (!isRunning() || isStale()) break;
 
 					try {
-						Debug('metamorphosis:app:consumer:debezium-mysql:verbose')('Message: %O', message);
+						Debug('metamorphosis:app:consumer:debezium-mysql:verbose')('Message: %o', message);
 
 						// Extract message value from Kafka message
 						const { value: messageValue } = message || {};
@@ -97,6 +97,7 @@ export class DebeziumMysqlConsumerService extends ConsumerService {
 						try {
 							parsedValue = this.registry ? await this.registry.decode(messageValue) : JSON.parse(messageValue.toString());
 						} catch (err) {
+							Debug('metamorphosis:app:consumer:debezium-mysql:debug')('Failed parsing message: %o', err);
 							resolveOffset(message.offset);
 							continue;
 						}