pubsub kafka: topic & config tweaks based on testing

LiberatorUSA · Mar 7, 2024 · 92f63d8 · 92f63d8
1 parent 8c32e99
commit 92f63d8
Show file tree

Hide file tree

Showing 2 changed files with 25 additions and 10 deletions.
diff --git a/plugins/PUBSUB/pubsubpluginKAFKA/src/pubsubpluginKAFKA_CKafkaPubSubClientTopic.cpp b/plugins/PUBSUB/pubsubpluginKAFKA/src/pubsubpluginKAFKA_CKafkaPubSubClientTopic.cpp
@@ -1785,7 +1785,12 @@ CKafkaPubSubClientTopic::offset_commit_cb( RdKafka::ErrorCode err
 
             // check our config to see how we want to handle this scenario
             Int64 requestTimeout = GetConsumerConfigSettingAsInt64( "request.timeout.ms", 3000 );
-            const CORE::CString& offsetResetSetting = GetConsumerConfigSetting( "auto.offset.reset", DefaultOffsetResetValue );            
+            CORE::CString offsetResetSetting = GetConsumerConfigSetting( "auto.offset.reset", DefaultOffsetResetValue );            
+            if ( offsetResetSetting.Lowercase() == "stored" )
+            {
+                GUCEF_WARNING_LOG( CORE::LOGLEVEL_NORMAL, "KafkaPubSubClientTopic: The \"auto.offset.reset\" setting is set to \"" + offsetResetSetting + " which is not correct for an invalid/no offset scenario. Overruling to " + DefaultOffsetResetValue );    
+                offsetResetSetting = DefaultOffsetResetValue; 
+            }
             GUCEF_SYSTEM_LOG( CORE::LOGLEVEL_NORMAL, "KafkaPubSubClientTopic: Using offset reset setting of \"" + offsetResetSetting + "\" with request timeout of " + CORE::ToString( requestTimeout ) );
 
             // Set the desired new offsets
@@ -1830,10 +1835,20 @@ CKafkaPubSubClientTopic::rebalance_cb( RdKafka::KafkaConsumer* consumer
         {
             if ( m_firstPartitionAssignment || RdKafka::Topic::OFFSET_INVALID == partitions[ i ]->offset() )
             {
+                CORE::Int64 startOffset = (CORE::Int64) RdKafka::Topic::OFFSET_INVALID;
                 Int64 requestTimeout = GetConsumerConfigSettingAsInt64( "request.timeout.ms", 3000 );
-                const CORE::CString& offsetResetSetting = GetConsumerConfigSetting( "auto.offset.reset", m_config.consumerModeStartOffset.IsNULLOrEmpty() ? DefaultOffsetResetValue : m_config.consumerModeStartOffset );
+                if ( RdKafka::Topic::OFFSET_INVALID == partitions[i]->offset() )
+                {
+                    startOffset = ConvertKafkaConsumerStartOffset( DefaultOffsetResetValue, partitions[ i ]->partition(), (Int32) requestTimeout );
+                }
+                else
+                {
+                    CORE::CString offsetResetValue = m_config.consumerModeStartOffset.IsNULLOrEmpty() ? DefaultOffsetResetValue : m_config.consumerModeStartOffset;
+                    offsetResetValue = GetConsumerConfigSetting( "auto.offset.reset", offsetResetValue ); 
+
+                    startOffset = ConvertKafkaConsumerStartOffset( offsetResetValue, partitions[ i ]->partition(), (Int32) requestTimeout );
+                }
 
-                CORE::Int64 startOffset = ConvertKafkaConsumerStartOffset( offsetResetSetting, partitions[ i ]->partition(), (Int32) requestTimeout );
                 partitions[ i ]->set_offset( startOffset );
                 m_consumerOffsets[ partitions[ i ]->partition() ] = startOffset;
             }

diff --git a/tools/pubsub2pubsub/config/examples/channel_templates/kafka2storage.json b/tools/pubsub2pubsub/config/examples/channel_templates/kafka2storage.json
@@ -119,7 +119,7 @@
                 "autoPersistMsgInterval": 1000,
                 "maxNrOfBookmarksToKeep": 10,
                 "CustomConfig": {
-                    "vfsRootPath": "InstallPath/bookmarks",
+                    "vfsRootPath": "DataPath/bookmarks",
                     "persistenceStructure": "{bookmarkNamespace}/clientType/{clientType}/topicName/{topicName}"
                 }
             }
@@ -155,8 +155,8 @@
                     "CustomConfig": {
                         "desiredNrOfBuffers": 3,
                         "desiredMinimalSerializedBlockSize": 52428800,
-                        "desiredMaxTimeToWaitToGrowSerializedBlockSizeInMs": 10000,
-                        "vfsStorageRootPath": "InstallPath/pubsub/storage/{pubsubIdPrefix}/{metricsFriendlyTopicName}",
+                        "desiredMaxTimeToWaitToGrowSerializedBlockSizeInMs": 240000,
+                        "vfsStorageRootPath": "DataPath/pubsub/storage/{pubsubIdPrefix}/{topicName}",
                         "vfsFileExtention": "pubsubcapture",
                         "encodeCodecFamily": "",
                         "encodeCodecName": "",
@@ -180,14 +180,14 @@
                         "treatEveryFullfilledRequestAsEODEvent": false,
                         "deleteContainersWithFullyAckdContent": false,
                         "moveContainersWithFullyAckdContent": true,
-                        "vfsStorageRootPathForFullyAckdContainers": "InstallPath/storage-sink/ackd",
+                        "vfsStorageRootPathForFullyAckdContainers": "DataPath/storage-sink/ackd",
                         "maxCompletedContainerRefsToRetain": 50,
                         "binarySerializerOptions": {
                             "msgDateTimeIncluded": true,
                             "msgDateTimeAsMsSinceUnixEpochInUtc": true,
                             "msgIdIncluded": true,
-                            "msgIndexIncluded": false,
-                            "msgPrimaryPayloadIncluded": false,
+                            "msgIndexIncluded": true,
+                            "msgPrimaryPayloadIncluded": true,
                             "msgKeyValuePairsIncluded": true,
                             "msgMetaDataKeyValuePairsIncluded": true,
                             "topicNameIncluded": false
@@ -254,7 +254,7 @@
                 "autoPersistMsgInterval": 1000,
                 "maxNrOfBookmarksToKeep": 10,
                 "CustomConfig": {
-                    "vfsRootPath": "InstallPath/pubsub/bookmarks",
+                    "vfsRootPath": "DataPath/pubsub/bookmarks",
                     "persistenceStructure": "{bookmarkNamespace}/clientType/{clientType}/topicName/{topicName}"
                 }
             }