Add getCachedBulletRecordProvider to BulletConfig

src/main/java/com/yahoo/bullet/common/BulletConfig.java

@@ -337,7 +337,6 @@ public class BulletConfig extends Config {
     public BulletConfig(String file) {
         super(file, DEFAULT_CONFIGURATION_NAME);
         VALIDATOR.validate(this);
-        provider = BulletRecordProvider.from(getAs(RECORD_PROVIDER_CLASS_NAME, String.class));
     }
 
     /**
@@ -346,19 +345,36 @@ public BulletConfig(String file) {
     public BulletConfig() {
         super(DEFAULT_CONFIGURATION_NAME);
         VALIDATOR.validate(this);
-        provider = BulletRecordProvider.from(getAs(RECORD_PROVIDER_CLASS_NAME, String.class));
     }
 
     /**
-     * Get the {@link BulletRecordProvider} stored in this BulletConfig instance. This BulletRecordProvider is
-     * created when this BulletConfig is constructed.
+     * Construct a {@link BulletRecordProvider} and store it in this BulletConfig instance.
      *
      * @return The BulletRecordProvider instance.
      */
     public BulletRecordProvider getBulletRecordProvider() {
+        provider = BulletRecordProvider.from(getAs(RECORD_PROVIDER_CLASS_NAME, String.class));
+        return provider;
+    }
+
+    /**
+     * Get the {@link BulletRecordProvider} stored in this BulletConfig instance, or construct and store one first if
+     * there is none.
+     *
+     * @return The BulletRecordProvider instance.
+     */
+    public BulletRecordProvider getCachedBulletRecordProvider() {
+        if (provider == null) {
+            return getBulletRecordProvider();
+        }
         return provider;
     }
 
+    /**
+     * Construct a {@link Schema} if configured.
+     *
+     * @return The Schema instance if configured; otherwise, null.
+     */
     public Schema getSchema() {
         String schemaFile = getAs(RECORD_SCHEMA_FILE_NAME, String.class);
         if (schemaFile == null) {

src/test/java/com/yahoo/bullet/common/BulletConfigTest.java

@@ -8,6 +8,8 @@
 import com.yahoo.bullet.querying.partitioning.MockPartitioner;
 import com.yahoo.bullet.record.BulletRecord;
 import com.yahoo.bullet.record.BulletRecordProvider;
+import com.yahoo.bullet.record.avro.TypedAvroBulletRecordProvider;
+import com.yahoo.bullet.record.simple.TypedSimpleBulletRecordProvider;
 import com.yahoo.bullet.result.Meta;
 import com.yahoo.bullet.result.Meta.Concept;
 import com.yahoo.bullet.typesystem.Type;
@@ -461,11 +463,13 @@ public void testGetBulletRecordProvider() {
         BulletConfig config = new BulletConfig();
         BulletRecordProvider providerA = config.getBulletRecordProvider();
         BulletRecordProvider providerB = config.getBulletRecordProvider();
-        Assert.assertEquals(providerA, providerB);
+
+        // Creates a new provider each time
+        Assert.assertNotEquals(providerA, providerB);
 
         // Ensure the provider generates new records each time
         BulletRecord recordA = providerA.getInstance();
-        BulletRecord recordB = providerB.getInstance();
+        BulletRecord recordB = providerA.getInstance();
 
         Assert.assertNotNull(recordA);
         Assert.assertNotNull(recordB);
@@ -475,6 +479,31 @@ public void testGetBulletRecordProvider() {
         Assert.assertTrue(recordA.typedGet("someField").isNull());
     }
 
+    @Test
+    public void testGetCachedBulletRecordProvider() {
+        BulletConfig config = new BulletConfig();
+        BulletRecordProvider providerA = config.getCachedBulletRecordProvider();
+        BulletRecordProvider providerB = config.getCachedBulletRecordProvider();
+
+        // Uses the same provider
+        Assert.assertSame(providerA, providerB);
+    }
+
+    @Test
+    public void testSettingDifferentBulletRecordProvider() {
+        BulletConfig config = new BulletConfig();
+
+        // Default record provider is TypedAvroBulletRecordProvider
+        Assert.assertTrue(config.getBulletRecordProvider() instanceof TypedAvroBulletRecordProvider);
+
+        config.set(BulletConfig.RECORD_PROVIDER_CLASS_NAME, TypedSimpleBulletRecordProvider.class.getName());
+
+        // Cached record provider doesn't change with new setting
+        Assert.assertTrue(config.getCachedBulletRecordProvider() instanceof TypedAvroBulletRecordProvider);
+
+        Assert.assertTrue(config.getBulletRecordProvider() instanceof TypedSimpleBulletRecordProvider);
+    }
+
     @Test(expectedExceptions = IllegalStateException.class)
     public void testEqualityPartitioningWithNoFieldsValidation() {
         BulletConfig config = new BulletConfig();